Skip to content
AlmasixAlmasix

Queues

Dispatching jobs

Section titled “Dispatching jobs”
from almasix.queue import Job, ShouldQueue, dispatch


class SendDigest(ShouldQueue, Job):
    tries = 3
    backoff = 10


    def __init__(self, user_id: int) -> None:
        self.user_id = user_id


    def handle(self) -> None:
        ...


await dispatch(SendDigest(1))
await SendDigest.dispatch(user_id=1)

Jobs without ShouldQueue (and without queue = True) run synchronously. Use dispatch_sync(job) to force in-process execution.

Drivers

Section titled “Drivers”
Connection Driver Notes
sync Immediate Default for tests/dev
database jobs / failed_jobs tables Call ensure_tables() or migrate
redis Redis lists + delayed ZSET Requires almasix[redis]; see Redis
config/queue.py
"redis": {
    "driver": "redis",
    "connection": "default",
    "queue": "queues",
},
QUEUE_CONNECTION=redis

Workers

Section titled “Workers”
Terminal window
python smith queue:work
python smith queue:listen
python smith queue:failed
python smith queue:retry {id}

Failed jobs call job.failed(exc) and report through the M8 exception Handler when available.

Section titled “Related”