gendesign/backend/app/workers/celery_app.py
lekss361 13025c81ee feat(cadastre): bulk_harvest_quarter Celery task + grid-walker + saga state (#168 PR3/5)
Add bulk cadastre harvest pipeline:
- services/cadastre/bulk_harvest.py: async harvest_quarter() orchestrator (4 phases)
  + 7 upsert helpers (parcels/buildings/constructions/oncs/enks/zouit/quarter_stats)
  using CAST(:x AS jsonb) pattern, begin_nested() SAVEPOINT per grid-walk upsert
- services/cadastre/grid_geometry.py: quarter_bbox_3857 (PostGIS ST_Extent)
  + generate_grid_click_points (15x15 = 225 sub-bbox grid)
- workers/tasks/scrape_cadastre.py: bulk_harvest_quarter_task (acks_late=True,
  dont_autoretry_for NspdBulkWafError), enqueue_cadastre_harvest, cleanup_zombies
- api/v1/admin_cadastre.py: 5 endpoints behind AdminTokenAuth — create/list/get/cancel/resume
- Tests: 13 service unit + 8 API tests
2026-05-15 13:03:35 +03:00

40 lines
1.2 KiB
Python

"""Celery app — single source of truth для Celery configuration.
Beat schedule build → app/workers/beat_schedule.py.
Worker lifecycle hooks (process_init, worker_ready) → app/workers/lifecycle.py.
"""
import logging
from celery import Celery
from app.core.config import settings
logger = logging.getLogger(__name__)
celery_app = Celery(
"gendesign",
broker=settings.redis_url,
backend=settings.redis_url,
include=[
"app.workers.tasks.scrape_kn",
"app.workers.tasks.refresh_analytics",
"app.workers.tasks.scrape_objective",
"app.workers.tasks.objective_etl",
"app.workers.tasks.nspd_geo",
"app.workers.tasks.nspd_sync",
"app.workers.tasks.poi_sync",
"app.workers.tasks.noise_sync",
"app.workers.tasks.pzz_sync",
"app.workers.tasks.scrape_cadastre",
],
)
celery_app.conf.timezone = "Europe/Moscow"
# Apply beat schedule (DB → fallback → hardcoded entries)
from app.workers.beat_schedule import build_beat_schedule # noqa: E402
celery_app.conf.beat_schedule = build_beat_schedule()
# Register lifecycle hooks (import for side-effect — signal decorator registration)
from app.workers import lifecycle # noqa: E402, F401