97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
# Роуты запуска задач синхронизации и просмотра истории sync-runs.
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, func
|
|
|
|
from ..deps import get_persistence
|
|
from ...storage.db import PersistenceService
|
|
from ...storage.models import SyncRun
|
|
from ...worker.celery_app import celery_app
|
|
from ...worker.tasks import sync_listing_task
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SyncListingRequest(BaseModel):
|
|
lane: str = "openlane_marketplace"
|
|
limit: int | None = None
|
|
only_new: bool | None = None
|
|
max_pages: int | None = None
|
|
concurrency: int | None = None
|
|
|
|
|
|
@router.post("/tasks/sync-listing")
|
|
def start_sync_listing(
|
|
body: SyncListingRequest,
|
|
):
|
|
result = sync_listing_task.apply_async(
|
|
kwargs={
|
|
"lane": body.lane,
|
|
"limit": body.limit,
|
|
"only_new": body.only_new,
|
|
"max_pages": body.max_pages,
|
|
"concurrency": body.concurrency,
|
|
},
|
|
queue="scraping",
|
|
)
|
|
|
|
return {
|
|
"task_id": result.id,
|
|
"status": "queued",
|
|
}
|
|
|
|
|
|
@router.get("/tasks/{task_id}")
|
|
def get_task_status(task_id: str):
|
|
result = celery_app.AsyncResult(task_id)
|
|
|
|
payload: dict = {
|
|
"task_id": task_id,
|
|
"state": result.state,
|
|
}
|
|
|
|
if result.successful():
|
|
payload["result"] = result.result
|
|
elif result.failed():
|
|
payload["error"] = str(result.result)
|
|
elif result.info is not None:
|
|
payload["meta"] = result.info
|
|
|
|
return payload
|
|
|
|
|
|
@router.get("/sync-runs")
|
|
def list_sync_runs(
|
|
page: int = Query(1, ge=1),
|
|
per_page: int = Query(20, ge=1, le=100),
|
|
persistence: PersistenceService = Depends(get_persistence),
|
|
):
|
|
with persistence.session_scope() as session:
|
|
total = session.execute(select(func.count(SyncRun.id))).scalar() or 0
|
|
offset = (page - 1) * per_page
|
|
runs = session.execute(
|
|
select(SyncRun).order_by(SyncRun.started_at.desc()).offset(offset).limit(per_page)
|
|
).scalars().all()
|
|
|
|
return {
|
|
"total": total,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"items": [
|
|
{
|
|
"id": r.id,
|
|
"started_at": r.started_at.isoformat() if r.started_at else None,
|
|
"finished_at": r.finished_at.isoformat() if r.finished_at else None,
|
|
"status": r.status,
|
|
"lane": r.lane,
|
|
"ids_fetched": r.ids_fetched,
|
|
"cars_upserted": r.cars_upserted,
|
|
"cars_failed": r.cars_failed,
|
|
"images_upserted": r.images_upserted,
|
|
"error_summary": r.error_summary,
|
|
}
|
|
for r in runs
|
|
],
|
|
}
|