32 lines
1006 B
Python
32 lines
1006 B
Python
"""Generation job maintenance tasks."""
|
|
|
|
import asyncio
|
|
|
|
from app.core.celery_app import celery_app
|
|
from app.core.logging import get_logger
|
|
from app.db.database import _get_session_factory
|
|
from app.services.generation_jobs import mark_stale_generation_jobs
|
|
from app.tasks.utils import run_with_disposed_engine
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
@celery_app.task
|
|
def prune_stale_generation_jobs_task():
|
|
"""Periodically mark stale running generation jobs as failed."""
|
|
|
|
logger.info("prune_stale_generation_jobs_task_started")
|
|
|
|
async def _run():
|
|
session_factory = _get_session_factory()
|
|
async with session_factory() as session:
|
|
return await mark_stale_generation_jobs(session)
|
|
|
|
try:
|
|
result = asyncio.run(run_with_disposed_engine(_run()))
|
|
logger.info("prune_stale_generation_jobs_task_completed", **result)
|
|
return result
|
|
except Exception as exc:
|
|
logger.error("prune_stale_generation_jobs_task_failed", error=str(exc))
|
|
raise
|