backend/app/workers/tasks.py

from __future__ import annotations

import asyncio

from app.workers.celery_app import celery_app
from app.workers.runtime import WorkerRuntime


@celery_app.task(bind=True, name="app.workers.tasks.process_job", max_retries=3)
def process_job(self, job_id: str) -> None:
    runtime = WorkerRuntime()
    try:
        asyncio.run(runtime.startup())
        asyncio.run(runtime.process_job(job_id, self.request.id))
    except Exception as exc:
        try:
            if self.request.retries < self.max_retries:
                asyncio.run(runtime.mark_retrying(job_id, str(exc)))
                raise self.retry(exc=exc, countdown=min(60, 10 * (self.request.retries + 1)))
            asyncio.run(runtime.mark_failed(job_id, str(exc)))
        finally:
            asyncio.run(runtime.shutdown())
        raise
    else:
        asyncio.run(runtime.shutdown())


@celery_app.task(name="app.workers.tasks.cleanup_expired_images")
def cleanup_expired_images() -> int:
    runtime = WorkerRuntime()
    try:
        asyncio.run(runtime.startup())
        return asyncio.run(runtime.cleanup_expired_images())
    finally:
        asyncio.run(runtime.shutdown())