214 lines
7.1 KiB
Python
214 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from celery.utils.log import get_task_logger
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import ProgrammingError
|
|
|
|
from iaai_sync_service.celery_app import app
|
|
from iaai_sync_service.client import IAAIClient
|
|
from iaai_sync_service.database import get_session
|
|
from iaai_sync_service.models import SYNC_RUN_TABLE_NAME, SyncRun
|
|
from iaai_sync_service.runtime_config import load_runtime_config
|
|
from iaai_sync_service.schema_bootstrap import ensure_schema
|
|
from iaai_sync_service.settings import get_settings
|
|
from iaai_sync_service.sync_engine import run_sync_once
|
|
|
|
logger = get_task_logger(__name__)
|
|
|
|
SYNC_LANE = "iaai_cars"
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def trim_error_summary(errors: list[str], max_len: int) -> str | None:
|
|
if not errors:
|
|
return None
|
|
joined = " | ".join(errors)
|
|
if len(joined) <= max_len:
|
|
return joined
|
|
return joined[: max_len - 3] + "..."
|
|
|
|
|
|
def _is_missing_sync_runs_error(exc: ProgrammingError) -> bool:
|
|
message = str(exc).lower()
|
|
return "does not exist" in message and SYNC_RUN_TABLE_NAME in message
|
|
|
|
|
|
def _create_sync_run(session, lane: str) -> SyncRun: # noqa: ANN001
|
|
run = SyncRun(
|
|
status="running",
|
|
lane=lane,
|
|
started_at=utc_now(),
|
|
ids_fetched=0,
|
|
cars_upserted=0,
|
|
cars_failed=0,
|
|
images_upserted=0,
|
|
)
|
|
session.add(run)
|
|
session.commit()
|
|
return run
|
|
|
|
|
|
@app.task(name="iaai.sync_cars_feed", queue="iaai_sync", bind=True)
|
|
def sync_cars_feed(self: Any) -> dict[str, Any]:
|
|
return execute_sync_job()
|
|
|
|
|
|
@app.task(name="iaai.keep_session_alive", queue="iaai_sync", bind=True)
|
|
def keep_session_alive(self: Any) -> dict[str, Any]:
|
|
return execute_keepalive_job()
|
|
|
|
|
|
def execute_sync_job() -> dict[str, Any]:
|
|
settings = get_settings()
|
|
session = get_session()
|
|
run_id: int | None = None
|
|
lock_acquired = False
|
|
|
|
try:
|
|
logger.info("IAAI sync job started base_url=%s", settings.iaai_base_url)
|
|
if settings.schema_bootstrap_enabled:
|
|
ensure_schema(session)
|
|
session.commit()
|
|
logger.info("Schema bootstrap is enabled and completed.")
|
|
else:
|
|
logger.info("Schema bootstrap is disabled.")
|
|
|
|
lock_acquired = bool(
|
|
session.execute(
|
|
text("SELECT pg_try_advisory_lock(:key)"),
|
|
{"key": settings.advisory_lock_key},
|
|
).scalar_one()
|
|
)
|
|
if not lock_acquired:
|
|
logger.info("IAAI sync skipped: advisory lock is already held.")
|
|
return {"status": "skipped_locked"}
|
|
|
|
try:
|
|
run = _create_sync_run(session=session, lane=SYNC_LANE)
|
|
except ProgrammingError as exc:
|
|
session.rollback()
|
|
if not _is_missing_sync_runs_error(exc):
|
|
raise
|
|
logger.warning(
|
|
"Table '%s' is missing. Running one-time schema bootstrap and retrying.",
|
|
SYNC_RUN_TABLE_NAME,
|
|
)
|
|
ensure_schema(session)
|
|
session.commit()
|
|
run = _create_sync_run(session=session, lane=SYNC_LANE)
|
|
run_id = run.id
|
|
logger.info("IAAI sync run created run_id=%s", run_id)
|
|
|
|
client = IAAIClient(settings=settings)
|
|
runtime_config = load_runtime_config(settings.sync_runtime_config_file, logger=logger)
|
|
logger.info("Loaded runtime config from '%s'", settings.sync_runtime_config_file)
|
|
|
|
stats, errors = run_sync_once(
|
|
session=session,
|
|
client=client,
|
|
settings=settings,
|
|
runtime_config=runtime_config,
|
|
)
|
|
|
|
run = session.get(SyncRun, run_id)
|
|
if run is None:
|
|
raise RuntimeError(f"sync_runs row disappeared for id={run_id}")
|
|
run.finished_at = utc_now()
|
|
run.ids_fetched = stats.ids_fetched
|
|
run.cars_upserted = stats.cars_upserted
|
|
run.cars_failed = stats.cars_failed
|
|
run.images_upserted = stats.images_upserted
|
|
run.status = "partial" if stats.cars_failed > 0 else "success"
|
|
run.error_summary = trim_error_summary(errors, settings.error_summary_max_len)
|
|
session.commit()
|
|
|
|
logger.info(
|
|
"IAAI sync completed status=%s ids=%s cars_upserted=%s cars_failed=%s images_upserted=%s",
|
|
run.status,
|
|
run.ids_fetched,
|
|
run.cars_upserted,
|
|
run.cars_failed,
|
|
run.images_upserted,
|
|
)
|
|
return {
|
|
"status": run.status,
|
|
"ids_fetched": run.ids_fetched,
|
|
"cars_upserted": run.cars_upserted,
|
|
"cars_failed": run.cars_failed,
|
|
"images_upserted": run.images_upserted,
|
|
"run_id": run.id,
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
session.rollback()
|
|
if run_id is not None:
|
|
failed = session.get(SyncRun, run_id)
|
|
if failed is not None:
|
|
failed.finished_at = utc_now()
|
|
failed.status = "failed"
|
|
failed.error_summary = trim_error_summary([str(exc)], settings.error_summary_max_len)
|
|
session.commit()
|
|
logger.exception("IAAI sync failed: %s", exc)
|
|
raise
|
|
finally:
|
|
if lock_acquired:
|
|
try:
|
|
session.execute(
|
|
text("SELECT pg_advisory_unlock(:key)"),
|
|
{"key": settings.advisory_lock_key},
|
|
)
|
|
session.commit()
|
|
except Exception: # noqa: BLE001
|
|
session.rollback()
|
|
logger.exception("Failed to release advisory lock.")
|
|
session.close()
|
|
|
|
|
|
def execute_keepalive_job() -> dict[str, Any]:
|
|
settings = get_settings()
|
|
session = get_session()
|
|
lock_acquired = False
|
|
|
|
try:
|
|
lock_acquired = bool(
|
|
session.execute(
|
|
text("SELECT pg_try_advisory_lock(:key)"),
|
|
{"key": settings.advisory_lock_key},
|
|
).scalar_one()
|
|
)
|
|
if not lock_acquired:
|
|
logger.info("IAAI keepalive skipped: advisory lock is already held.")
|
|
return {"status": "skipped_locked"}
|
|
|
|
logger.info("IAAI keepalive job started base_url=%s", settings.iaai_base_url)
|
|
client = IAAIClient(settings=settings)
|
|
result = client.keep_session_alive()
|
|
logger.info(
|
|
"IAAI keepalive completed scope=%s marker_present=%s result_count=%s elapsed_ms=%s",
|
|
result.get("scope"),
|
|
result.get("marker_present"),
|
|
result.get("result_count"),
|
|
result.get("elapsed_ms"),
|
|
)
|
|
return {"status": "success", **result}
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("IAAI keepalive failed: %s", exc)
|
|
raise
|
|
finally:
|
|
if lock_acquired:
|
|
try:
|
|
session.execute(
|
|
text("SELECT pg_advisory_unlock(:key)"),
|
|
{"key": settings.advisory_lock_key},
|
|
)
|
|
session.commit()
|
|
except Exception: # noqa: BLE001
|
|
session.rollback()
|
|
logger.exception("Failed to release advisory lock for keepalive.")
|
|
session.close()
|