Skip to content

workers.arq

arq

ARQ workers infrastructure — base classes, config, task utilities.

Classes

BaseArqService

Async ARQ client for use inside worker tasks.

For the Django producer-side client see codex_platform.adapters.arq.client.BaseArqClient.

Example::

service = BaseArqService(redis_settings)
await service.init()
await service.enqueue_job("my_task", arg1, arg2)
await service.close()
Source code in src/codex_platform/workers/arq/base.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class BaseArqService:
    """Async ARQ client for use inside worker tasks.

    For the Django producer-side client see
    ``codex_platform.adapters.arq.client.BaseArqClient``.

    Example::

        service = BaseArqService(redis_settings)
        await service.init()
        await service.enqueue_job("my_task", arg1, arg2)
        await service.close()
    """

    def __init__(self, redis_settings: RedisSettings) -> None:
        """Initialize the service with Redis connection settings.

        Args:
            redis_settings: ARQ ``RedisSettings`` instance.
        """
        self.pool: ArqRedis | None = None
        self.redis_settings = redis_settings

    async def init(self) -> None:
        """Open the ARQ connection pool. No-op if already initialized.

        Raises:
            Exception: If the Redis connection cannot be established.
        """
        if not self.pool:
            try:
                self.pool = await create_pool(self.redis_settings)
                log.debug("BaseArqService | init | status=success")
            except Exception:
                log.exception("BaseArqService | init | status=failed")
                raise

    async def close(self) -> None:
        """Close the ARQ connection pool gracefully."""
        if self.pool:
            try:
                await self.pool.close()
                log.debug("BaseArqService | close | status=success")
            except Exception:
                log.exception("BaseArqService | close | status=failed")

    async def enqueue_job(self, function: str, *args: Any, **kwargs: Any) -> Any | None:
        """Enqueue a background job. Initializes the pool on first call.

        Args:
            function: ARQ worker function name.
            *args: Positional arguments forwarded to the function.
            **kwargs: Keyword arguments forwarded to the function.

        Returns:
            ARQ ``Job`` handle, or ``None`` if enqueueing failed.
        """
        if not self.pool:
            await self.init()
        if self.pool:
            try:
                job = await self.pool.enqueue_job(function, *args, **kwargs)
                log.debug(
                    "BaseArqService | enqueue_job | function=%s | job_id=%s",
                    function,
                    job.job_id if job else "None",
                )
                return job
            except Exception:
                log.exception("BaseArqService | enqueue_job | function=%s | status=failed", function)
                return None
        return None
Functions
__init__(redis_settings)

Initialize the service with Redis connection settings.

Parameters:

Name Type Description Default
redis_settings RedisSettings

ARQ RedisSettings instance.

 required
Source code in src/codex_platform/workers/arq/base.py 
40
41
42
43
44
45
46
47
def __init__(self, redis_settings: RedisSettings) -> None:
    """Initialize the service with Redis connection settings.

    Args:
        redis_settings: ARQ ``RedisSettings`` instance.
    """
    self.pool: ArqRedis | None = None
    self.redis_settings = redis_settings
init() async

Open the ARQ connection pool. No-op if already initialized.

Raises:

Type Description
Exception

If the Redis connection cannot be established.
Source code in src/codex_platform/workers/arq/base.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
async def init(self) -> None:
    """Open the ARQ connection pool. No-op if already initialized.

    Raises:
        Exception: If the Redis connection cannot be established.
    """
    if not self.pool:
        try:
            self.pool = await create_pool(self.redis_settings)
            log.debug("BaseArqService | init | status=success")
        except Exception:
            log.exception("BaseArqService | init | status=failed")
            raise
close() async

Close the ARQ connection pool gracefully.

Source code in src/codex_platform/workers/arq/base.py 
63
64
65
66
67
68
69
70
async def close(self) -> None:
    """Close the ARQ connection pool gracefully."""
    if self.pool:
        try:
            await self.pool.close()
            log.debug("BaseArqService | close | status=success")
        except Exception:
            log.exception("BaseArqService | close | status=failed")
enqueue_job(function, *args, **kwargs) async

Enqueue a background job. Initializes the pool on first call.

Parameters:

Name Type Description Default
function str

ARQ worker function name.

 required
*args Any

Positional arguments forwarded to the function.

 ()
**kwargs Any

Keyword arguments forwarded to the function.

 {}

Returns:

Type Description
Any | None

ARQ Job handle, or None if enqueueing failed.
Source code in src/codex_platform/workers/arq/base.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def enqueue_job(self, function: str, *args: Any, **kwargs: Any) -> Any | None:
    """Enqueue a background job. Initializes the pool on first call.

    Args:
        function: ARQ worker function name.
        *args: Positional arguments forwarded to the function.
        **kwargs: Keyword arguments forwarded to the function.

    Returns:
        ARQ ``Job`` handle, or ``None`` if enqueueing failed.
    """
    if not self.pool:
        await self.init()
    if self.pool:
        try:
            job = await self.pool.enqueue_job(function, *args, **kwargs)
            log.debug(
                "BaseArqService | enqueue_job | function=%s | job_id=%s",
                function,
                job.job_id if job else "None",
            )
            return job
        except Exception:
            log.exception("BaseArqService | enqueue_job | function=%s | status=failed", function)
            return None
    return None

BaseArqWorkerSettings

Base ARQ WorkerSettings. Subclass and override fields in your workers module.

Example::

class MyWorkerSettings(BaseArqWorkerSettings):
    redis_settings = RedisSettings(host="localhost", port=6379)
    functions = [my_task1, my_task2] + CORE_FUNCTIONS
    on_startup = my_startup
    on_shutdown = my_shutdown
Source code in src/codex_platform/workers/arq/base.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class BaseArqWorkerSettings:
    """Base ARQ ``WorkerSettings``. Subclass and override fields in your workers module.

    Example::

        class MyWorkerSettings(BaseArqWorkerSettings):
            redis_settings = RedisSettings(host="localhost", port=6379)
            functions = [my_task1, my_task2] + CORE_FUNCTIONS
            on_startup = my_startup
            on_shutdown = my_shutdown
    """

    max_jobs: int = 20
    job_timeout: int = 60
    keep_result: int = 60
    max_retries: int = 5
    retry_delay: int = 10

    on_startup = base_startup
    on_shutdown = base_shutdown

BaseWorkerConfig

Bases: BaseCommonSettings

Base configuration for ARQ workers — provides SMTP and ARQ settings.

Maps standard Django .env variable names (EMAIL_HOST, etc.) to internal SMTP_-prefixed fields for clarity.

Extend this class in your project to add vendor-specific fields (Twilio, Seven.io, SendGrid, etc.).

Example::

class MyWorkerConfig(BaseWorkerConfig):
    SEVEN_IO_API_KEY: str | None = None
    TEMPLATES_DIR: str = "src/workers/templates"

    class Config:
        env_file = ".env"
Source code in src/codex_platform/workers/arq/config.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class BaseWorkerConfig(BaseCommonSettings):
    """Base configuration for ARQ workers — provides SMTP and ARQ settings.

    Maps standard Django ``.env`` variable names (``EMAIL_HOST``, etc.) to
    internal ``SMTP_``-prefixed fields for clarity.

    Extend this class in your project to add vendor-specific fields
    (Twilio, Seven.io, SendGrid, etc.).

    Example::

        class MyWorkerConfig(BaseWorkerConfig):
            SEVEN_IO_API_KEY: str | None = None
            TEMPLATES_DIR: str = "src/workers/templates"

            class Config:
                env_file = ".env"
    """

    # --- Email (SMTP) — Django env name mapping ---
    SMTP_HOST: str = Field(default="localhost", alias="EMAIL_HOST")
    SMTP_PORT: int = Field(default=465, alias="EMAIL_PORT")
    SMTP_USER: str | None = Field(default=None, alias="EMAIL_HOST_USER")
    SMTP_PASSWORD: str | None = Field(default=None, alias="EMAIL_HOST_PASSWORD")
    SMTP_FROM_EMAIL: str | None = Field(default=None, alias="DEFAULT_FROM_EMAIL")
    SMTP_USE_TLS: bool = Field(default=False, alias="EMAIL_USE_TLS")
    SMTP_USE_SSL: bool = Field(default=True, alias="EMAIL_USE_SSL")

    # --- ARQ Configuration ---
    arq_max_jobs: int = 10
    arq_job_timeout: int = 60
    arq_keep_result: int = 60

    @property
    def arq_redis_settings(self) -> Any:
        """Build and return an ARQ ``RedisSettings`` object from the base Redis config."""
        from arq.connections import RedisSettings

        return RedisSettings(
            host=self.redis_host,
            port=self.redis_port,
            password=self.redis_password,
        )
Attributes
arq_redis_settings property

Build and return an ARQ RedisSettings object from the base Redis config.

Functions

base_shutdown(ctx) async

Base shutdown hook. Removes health check file. Extend in your workers.

Source code in src/codex_platform/workers/arq/base.py 
111
112
113
114
async def base_shutdown(ctx: dict[str, Any]) -> None:
    """Base shutdown hook. Removes health check file. Extend in your workers."""
    HEALTH_FILE.unlink(missing_ok=True)
    log.info("ArqWorker | shutdown")

base_startup(ctx) async

Base startup hook. Creates health check file. Extend in your workers.

Source code in src/codex_platform/workers/arq/base.py 
105
106
107
108
async def base_startup(ctx: dict[str, Any]) -> None:
    """Base startup hook. Creates health check file. Extend in your workers."""
    HEALTH_FILE.touch()
    log.info("ArqWorker | startup | health_file=%s", HEALTH_FILE)

requeue_to_stream(ctx, stream_name, payload) async

Re-enqueue a failed message back into a Redis Stream for retry processing.

After 5 retries the message is moved to a Dead Letter Queue (DLQ) instead of being dropped. DLQ key: {stream_name}:dlq.

Parameters:

Name Type Description Default
ctx dict[str, Any]

ARQ worker context dict. Must contain a "stream_manager" key.

 required
stream_name str

Target Redis Stream name.

 required
payload dict[str, Any]

Original message payload. _retries counter is incremented in-place.

 required
Source code in src/codex_platform/workers/arq/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
async def requeue_to_stream(ctx: dict[str, Any], stream_name: str, payload: dict[str, Any]) -> None:
    """Re-enqueue a failed message back into a Redis Stream for retry processing.

    After 5 retries the message is moved to a Dead Letter Queue (DLQ) instead of
    being dropped. DLQ key: ``{stream_name}:dlq``.

    Args:
        ctx:         ARQ worker context dict. Must contain a ``"stream_manager"`` key.
        stream_name: Target Redis Stream name.
        payload:     Original message payload. ``_retries`` counter is incremented in-place.
    """
    sm = ctx.get("stream_manager")
    if not sm:
        log.error("requeue_to_stream | stream_manager not in context")
        return
    retries = int(payload.get("_retries", 0)) + 1
    if retries > 5:
        dlq_name = f"{stream_name}:dlq"
        payload["_failed_at"] = datetime.now(UTC).isoformat()
        payload["_original_stream"] = stream_name
        await sm.add_event(dlq_name, payload)
        log.error("requeue_to_stream | max retries reached | moved to DLQ '%s'", dlq_name)
        return
    payload["_retries"] = str(retries)
    await sm.add_event(stream_name, payload)
    log.info("requeue_to_stream | requeued to '%s' retry=%d", stream_name, retries)

arq_task(*, retry_backoff=30, max_retries=5, log_args=False)

Decorator that adds standard error handling, logging, and retry to ARQ tasks.

Parameters:

Name Type Description Default
retry_backoff int

Base delay in seconds. Actual delay = job_try * retry_backoff.

 30
max_retries int

Maximum number of retries before giving up.

 5
log_args bool

If True, log task arguments (be careful with PII).

 False

Wrapped function receives ctx as first arg (standard ARQ convention). On exception: - If job_try < max_retries → raises Retry(defer=job_try * retry_backoff) - If job_try >= max_retries → logs error and returns None (task dropped)

Source code in src/codex_platform/workers/arq/task_utils.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def arq_task(
    *,
    retry_backoff: int = 30,
    max_retries: int = 5,
    log_args: bool = False,
) -> Any:
    """
    Decorator that adds standard error handling, logging, and retry to ARQ tasks.

    Args:
        retry_backoff: Base delay in seconds. Actual delay = job_try * retry_backoff.
        max_retries: Maximum number of retries before giving up.
        log_args: If True, log task arguments (be careful with PII).

    Wrapped function receives ctx as first arg (standard ARQ convention).
    On exception:
        - If job_try < max_retries → raises Retry(defer=job_try * retry_backoff)
        - If job_try >= max_retries → logs error and returns None (task dropped)
    """

    def decorator(fn: Any) -> Any:
        @wraps(fn)
        async def wrapper(ctx: dict[str, Any], *args: Any, **kwargs: Any) -> Any:
            task_name = fn.__name__
            job_try = ctx.get("job_try", 1)

            if log_args:
                log.info("%s | start | try=%d | args=%s kwargs=%s", task_name, job_try, args, kwargs)
            else:
                log.info("%s | start | try=%d", task_name, job_try)

            try:
                result = await fn(ctx, *args, **kwargs)
                log.info("%s | success | try=%d", task_name, job_try)
                return result
            except Retry:
                raise
            except Exception as e:
                if job_try < max_retries:
                    defer = job_try * retry_backoff
                    log.warning("%s | retry in %ds | try=%d | error=%s", task_name, defer, job_try, e)
                    raise Retry(defer=defer) from e
                log.exception("%s | max retries (%d) reached | giving up", task_name, max_retries)
                return None

        wrapper.__qualname__ = fn.__qualname__
        return wrapper

    return decorator