Add job start and end handlers to worker functions, integrating structlog for enhanced logging context. Update worker settings to include new handlers.

Author: sri-dhurkesh

src/app/core/worker/functions.py

@@ -1,13 +1,12 @@
 import asyncio
 import logging
 
+import structlog
 import uvloop
 from arq.worker import Worker
 
 asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
-logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-
 
 # -------- background tasks --------
 async def sample_background_task(ctx: Worker, name: str) -> str:
@@ -22,3 +21,13 @@ async def startup(ctx: Worker) -> None:
 
 async def shutdown(ctx: Worker) -> None:
     logging.info("Worker end")
+
+
+async def on_job_start(ctx: Worker) -> None:
+    structlog.contextvars.bind_contextvars(job_id=ctx["job_id"])
+    logging.info("Job Started")
+
+
+async def on_job_end(ctx: Worker) -> None:
+    logging.info("Job Competed")
+    structlog.contextvars.clear_contextvars()

src/app/core/worker/settings.py

@@ -1,7 +1,14 @@
+import asyncio
+from typing import cast
+
+from arq.cli import watch_reload
 from arq.connections import RedisSettings
+from arq.typing import WorkerSettingsType
+from arq.worker import check_health, run_worker
 
 from ...core.config import settings
-from .functions import sample_background_task, shutdown, startup
+from ...core.logger import logging  # noqa: F401
+from .functions import on_job_end, on_job_start, sample_background_task, shutdown, startup
 
 REDIS_QUEUE_HOST = settings.REDIS_QUEUE_HOST
 REDIS_QUEUE_PORT = settings.REDIS_QUEUE_PORT
@@ -12,4 +19,24 @@ class WorkerSettings:
     redis_settings = RedisSettings(host=REDIS_QUEUE_HOST, port=REDIS_QUEUE_PORT)
     on_startup = startup
     on_shutdown = shutdown
+    on_job_start = on_job_start
+    on_job_end = on_job_end
     handle_signals = False
+
+
+def start_arq_service(check: bool = False, burst: int | None = None, watch: str | None = None):
+    worker_settings_ = cast("WorkerSettingsType", WorkerSettings)
+
+    if check:
+        exit(check_health(worker_settings_))
+    else:
+        kwargs = {} if burst is None else {"burst": burst}
+        if watch:
+            asyncio.run(watch_reload(watch, worker_settings_))
+        else:
+            run_worker(worker_settings_, **kwargs)
+
+
+if __name__ == "__main__":
+    start_arq_service()
+    # python -m src.app.core.worker.settings