38 lines
1.0 KiB
Python
38 lines
1.0 KiB
Python
|
"""arq worker process entrypoint.
|
||
|
|
|
||
|
|
Run with: `arq app.workers.worker.WorkerSettings`
|
||
|
|
|
||
|
|
Workers scale independently of the API (section 1) — run as many worker
|
||
|
|
processes/containers as needed; arq's Redis-backed queue coordinates them so
|
||
|
|
each job is picked up by exactly one worker.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from arq.connections import RedisSettings
|
||
|
|
|
||
|
|
from app.core.config import get_settings
|
||
|
|
from app.core.logging import configure_logging, get_logger
|
||
|
|
from app.workers.tasks import ingest_movie_task, ingest_person_task
|
||
|
|
|
||
|
|
logger = get_logger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
async def startup(ctx: dict) -> None:
|
||
|
|
configure_logging()
|
||
|
|
logger.info("worker_started")
|
||
|
|
|
||
|
|
|
||
|
|
async def shutdown(ctx: dict) -> None:
|
||
|
|
logger.info("worker_stopped")
|
||
|
|
|
||
|
|
|
||
|
|
class WorkerSettings:
|
||
|
|
functions = [ingest_movie_task, ingest_person_task]
|
||
|
|
on_startup = startup
|
||
|
|
on_shutdown = shutdown
|
||
|
|
redis_settings = RedisSettings.from_dsn(get_settings().REDIS_URL)
|
||
|
|
max_jobs = get_settings().WORKER_CONCURRENCY
|
||
|
|
job_timeout = 300
|
||
|
|
max_tries = 5
|