Agent Zero
727d86614e
P0 (7): Auth-bypass removed, migrations fixed, plugin-upload disabled, RLS FORCE+WITH CHECK, plugin double-registration fixed, persistent volume, domain removed P1 (11): User/tenant model, Redis centralized, worker separated, transactional outbox, XSS fixed, DMS chunked streaming, permissions unified, password reset, metrics secured, config/docs fixed, cross-tenant FK P2 (4): Contact model normalized, cross-imports reduced 94%, commands+state machines for contacts/dms/mail/calendar, SPA path-traversal 8 new migrations, 99 unit tests, 13 commands, 8 contracts, 72 files changed
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""ARQ job queue integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from arq import create_pool
|
|
from arq.connections import RedisSettings, ArqRedis
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Global ARQ pool singleton ────────────────────────────────────────────────
|
|
_job_pool: ArqRedis | None = None
|
|
|
|
|
|
async def init_job_pool() -> ArqRedis:
|
|
"""Create and store the global ARQ job pool.
|
|
|
|
Called once during app lifespan startup so every subsequent enqueue
|
|
reuses the same connection instead of opening a new one per call.
|
|
"""
|
|
global _job_pool
|
|
if _job_pool is not None:
|
|
logger.warning("init_job_pool() called but pool already initialized")
|
|
return _job_pool
|
|
settings = get_settings()
|
|
redis_settings = RedisSettings.from_dsn(settings.redis_url)
|
|
_job_pool = await create_pool(redis_settings)
|
|
logger.info("Global ARQ job pool initialized")
|
|
return _job_pool
|
|
|
|
|
|
async def close_job_pool() -> None:
|
|
"""Close the global ARQ job pool. Called during app lifespan shutdown."""
|
|
global _job_pool
|
|
if _job_pool is not None:
|
|
await _job_pool.close()
|
|
_job_pool = None
|
|
logger.info("Global ARQ job pool closed")
|
|
|
|
|
|
async def get_job_pool() -> ArqRedis:
|
|
"""Return the global ARQ job pool singleton.
|
|
|
|
If init_job_pool() has not been called yet (e.g. during testing),
|
|
a new pool is created lazily so callers always get a working connection.
|
|
"""
|
|
global _job_pool
|
|
if _job_pool is None:
|
|
settings = get_settings()
|
|
redis_settings = RedisSettings.from_dsn(settings.redis_url)
|
|
_job_pool = await create_pool(redis_settings)
|
|
logger.debug("ARQ job pool created lazily (init_job_pool not called)")
|
|
return _job_pool
|
|
|
|
|
|
async def enqueue_job(job_name: str, *args: Any, **kwargs: Any) -> str | None:
|
|
"""Enqueue a background job. Returns job_id or None."""
|
|
pool = await get_job_pool()
|
|
job = await pool.enqueue_job(job_name, *args, **kwargs)
|
|
if job:
|
|
return job.job_id
|
|
return None
|
|
|
|
|
|
async def get_job_status(job_id: str) -> dict[str, Any] | None:
|
|
"""Get the status of a background job."""
|
|
pool = await get_job_pool()
|
|
job_info = await pool.job_info(job_id)
|
|
if job_info is None:
|
|
return None
|
|
return {
|
|
"job_id": job_id,
|
|
"status": str(job_info.status),
|
|
"result": job_info.result,
|
|
"enqueue_time": job_info.enqueue_time.isoformat() if job_info.enqueue_time else None,
|
|
"start_time": job_info.start_time.isoformat() if job_info.start_time else None,
|
|
"finish_time": job_info.finish_time.isoformat() if job_info.finish_time else None,
|
|
}
|