streaQ Distributed Task Queue

You can define tasks on separate Worker instances and combine them using worker.include(other_worker). This allows for grouping tasks logically.

Accessing Context in Modular Tasks When a task is defined on a separate worker instance (e.g., other), it can still access the dependencies of the main worker that includes it. To maintain type hints, define the worker with the expected context type: other = Worker[WorkerContext]().

Warning: Accessing the worker inside a task Do not use the global worker variable (e.g., other.abort_by_id(...)) inside a task, as other might not be the actual running worker. Instead, use the reference attached to the task object: task_name.worker.abort_by_id(...).

# other.py
from streaq import Worker
from typing import NamedTuple
from httpx import AsyncClient

class WorkerContext(NamedTuple):
    client: AsyncClient

other = Worker[WorkerContext]()

@other.task
async def make_request() -> None:
    await other.context.client.get("...")

@other.task
async def abort_foobar(task_id: str) -> bool:
    # Use the task's worker reference, NOT the global 'other'
    return await make_request.worker.abort_by_id(task_id)

# main.py
from streaq import Worker
from other import other, make_request

worker = Worker()
worker.include(other)

@worker.task
async def barfoo() -> bool: ...

async def main():
    async with worker:
        await make_request.enqueue()
        await barfoo.enqueue()

Middleware allows you to wrap task execution to implement cross-cutting concerns like observability, logging, or exception handling.

Middleware is implemented as a higher-order function that accepts a ReturnCoroutine (the task to be executed) and returns a new ReturnCoroutine (the wrapper). Because they are structured as wrapped functions, you can:

1. Execute code before the task starts.
2. Execute code after the task completes.
3. Access and modify the task's arguments (*args, **kwargs).
4. Access and modify the task's return value.
5. Access the task context via the RegisteredMiddleware.context attribute (e.g., middleware_name.context.task_id).

For automated distributed tracing, use the opentelemetry-instrumentation-streaq package instead of manual middleware.

import time
from typing import Any
from streaq.types import ReturnCoroutine

@worker.middleware
def timer(task: ReturnCoroutine) -> ReturnCoroutine:
    async def wrapper(*args, **kwargs) -> Any:
        start_time = time.perf_counter()
        result = await task(*args, **kwargs)
        print(f"Executed task {timer.context.task_id} in {time.perf_counter() - start_time:.3f}s")
        return result

    return wrapper

When interacting with tasks, streaQ provides two main data structures:

TaskInfo: Contains metadata about the task's lifecycle.

• fn_name: Name of the function.
• enqueue_time: Timestamp of enqueuing.
• tries: Number of attempts made.
• scheduled: The scheduled execution time.
• dependencies / dependents: Sets representing the task's position in the dependency graph.

TaskResult: Contains the outcome of the task execution.

• fn_name: Name of the function.
• success: Boolean indicating if the task completed without error.
• start_time / finish_time: Execution timestamps.
• tries: Total attempts.
• worker_id: The ID of the worker that processed the task.
• _result: The actual return value of the function.

# Example TaskInfo output
TaskInfo(fn_name='sleeper', enqueue_time=1751508876961, tries=0, scheduled=datetime.datetime(2025, 7, 3, 2, 14, 39, 961000, tzinfo=datetime.timezone.utc), dependencies=set(), dependents=set())

# Example TaskResult output
TaskResult(fn_name='sleeper', enqueue_time=1751508876961, success=True, start_time=1751508880500, finish_time=1751508881503, tries=1, worker_id='ca5bd9eb', _result=1)

streaQ uses a "pessimistic execution" (at-least-once) model. Tasks are only removed from the queue after they succeed or fail. If a worker shuts down mid-task, the task remains in the queue for another worker to pick up.

Key implications:

• Idempotency: Tasks should be designed to be safe if called multiple times. Use database transactions, idempotency keys, or Redis to ensure non-repeatable work isn't performed twice.
• Opting out: You can disable retries for a specific task by setting max_tries=1 in the @worker.task decorator.
• Liveness: Worker.idle_timeout determines how quickly stuck tasks are retried.

Exception handling:

• StreaqRetry: Triggers a retry (potentially with delay).
• asyncio.CancelledError or trio.Cancelled: Results in failure if aborted by the user, or a retry if the worker shuts down unexpectedly.
• Any other exception: Results in task failure.