SAQ (Simple Async Queue)

SAQ (Simple Async Queue) is a performant job queueing framework built on top of asyncio and redis. It is designed for processing background jobs such as scheduling emails, executing long-running queries, or performing expensive data analysis using workers.

Key characteristics:

• Async-native: Built on asyncio, making it significantly faster than RQ for async jobs.
• Low overhead: Offers lower overhead than traditional alternatives even for synchronous jobs.
• Redis-based: Requires redis-py >= 4.2.

SAQ is a high-performance task queue inspired by ARQ, designed for lower latency and better monitoring. Key advantages over ARQ include:

• Lower Latency: SAQ leverages Redis BLMOVE or RPOPLPUSH and NOTIFY to avoid polling, achieving delays of < 5ms (compared to ARQ's default 0.5s polling).
• Performance: SAQ is up to 8x faster than ARQ in certain workflows.
• Monitoring: Includes a web interface for monitoring queues and workers.
• Reliability: Features a heartbeat monitor for abandoned jobs, robust failure handling (including stack trace storage), and mechanisms to sweep stuck jobs.
• Job Lifecycle: Supports before and after job hooks and distinguishes between cancelled jobs (e.g., during machine redeployments) and failed jobs.
• Scalability: Designed to easily run multiple workers to leverage multiple CPU cores.

In SAQ, every job function is passed a ctx dictionary. This dictionary is the primary way to share state (like database connections or shared clients) between the worker's lifecycle hooks (startup, shutdown) and the individual jobs.

1. Startup: Use the startup hook to create a resource and store it in ctx (e.g., ctx['db'] = connection).
2. Job Execution: The job function receives ctx and can access the resource (e.g., await ctx['db'].query(...)).
3. Shutdown: Use the shutdown hook to close resources stored in ctx.

async def startup(ctx):
    ctx["db"] = await connect_db()

async def my_job(ctx, *args, **kwargs):
    # Access the shared resource via ctx
    await ctx["db"].execute("SELECT 1")

async def shutdown(ctx):
    await ctx["db"].disconnect()

In SAQ, it is important to distinguish between a Task and a Job:

• Task: The blueprint. A Python function you define in your codebase containing the logic to be executed.
• Job: A specific execution of a Task. Created when you enqueue a task with specific arguments. One Task definition can generate millions of individual Jobs.

Every task receives a Context object as its first argument. This dictionary-like object provides runtime information about the job:

• ctx['job']: The Job instance being executed (provides access to key, attempts, and meta).
• ctx['worker']: The Worker instance processing the job.
• ctx['queue']: The Queue the job was pulled from.
• ctx['exception']: If the task is being retried, this holds the exception from the previous failed attempt.

A task is a Python function that must follow these rules:

1. The first argument must be a saq.types.Context object.
2. Subsequent arguments must be keyword arguments (**kwargs).
3. The return value must be JSON serializable.

While both synchronous and asynchronous functions work, async def is recommended for non-blocking capabilities.

import asyncio
from saq.types import Context

# A task is a function that takes a context and keyword arguments.
async def send_welcome_email(ctx: Context, *, user_id: int) -> dict:
    print(f"Attempting to send email to user {user_id}...")
    await asyncio.sleep(1)  # Simulate network I/O
    return {"status": "sent", "user_id": user_id}

Install SAQ using pip with the appropriate extras depending on your backend and requirements:

• Redis backend (minimal): pip install saq[redis]
• Postgres backend (minimal): pip install saq[postgres]
• Web UI + hiredis (recommended for performance and monitoring): pip install saq[web,hiredis]

# minimal install for redis
pip install saq[redis]

# minimal install for postgres
pip install saq[postgres]

# web + hiredis
pip install saq[web,hiredis]

If you want to integrate the SAQ monitoring interface into an existing web service, use the saq_web function from saq.web.starlette. This function returns a Starlette application instance that can be mounted as a sub-application.

When mounting, you must provide a list of queues to monitor via the queues argument.

from saq.web.starlette import saq_web
from starlette.routing import Mount

# Example: Mounting the UI at the '/monitor' path
routes = [
    ...
    Mount("/monitor", saq_web("/monitor", queues=all_the_queues_list))
]

A Worker must be explicitly aware of the task functions it is allowed to execute. You register tasks by passing a list of function objects to the Worker constructor. SAQ identifies tasks using their qualified name (e.g., mymodule.send_welcome_email).

from saq.worker import Worker
from saq.queue import Queue
from .tasks import send_welcome_email

queue = Queue.from_url("redis://localhost")

# The worker needs to know about the task functions it can execute.
worker = Worker(queue=queue, functions=[send_welcome_email])

To run tasks on a recurring schedule, define a CronJob using standard cron syntax and pass it to the Worker via the cron_jobs argument.

from saq.job import CronJob

async def cleanup_task(ctx):
    print("Performing nightly cleanup...")

cron_jobs = [
    # Run every day at midnight
    CronJob(cleanup_task, cron="0 0 * * *")
]

worker = Worker(queue=queue, functions=[cleanup_task], cron_jobs=cron_jobs)

You can compare the performance of saq against other task queues like arq and rq by installing the respective packages and running the benchmarks/simple.py script. The benchmarks measure enqueue time, noop (no-operation) time, and sleep time for a workload of N=1000.

# Benchmark saq (Redis)
pip install saq && python benchmarks/simple.py saq

# Benchmark saq (Postgres)
pip install saq && python benchmarks/simple.py saq_pg

# Benchmark arq
pip install arq && python benchmarks/simple.py arq

# Benchmark rq
pip install rq && python benchmarks/simple.py rq

You can launch the SAQ monitoring UI directly alongside your worker process using the --web flag. By default, the UI is served on port 8080. To use a different port, provide the --port <portnum> flag.

# Run with default port 8080
saq examples.simple.settings --web

# Run on a custom port (e.g., 7000)
saq examples.simple.settings --web --port 7000