Resource System

The resource system gives tasks clean access to external dependencies — database connections, HTTP clients, cloud clients — without passing live objects through the queue. It operates in three layers that together solve a fundamental distributed systems problem: task arguments must be serializable (able to be turned into bytes so they can cross the process boundary into the worker), but most real-world dependencies are not.

Coming from Celery (which has no first-class DI), resources replace the module-level globals you'd otherwise create — a single db = create_engine(...) at import time — with per-worker instances that flexiq initializes, scopes, and tears down for you.

Two terms come up throughout this section: a recipe is the small, serializable dict a proxy handler extracts from a live object (e.g. a file path and mode) so the worker can rebuild it later, and a DI marker is the placeholder that replaces a redirected argument (like a database session) so the worker knows which named resource to inject in its place.

Layer 1 — Argument Interception classifies each value passed to .delay() before serialization. Database sessions become DI markers, file handles become recipes, safe primitives pass through unchanged, and non-serializable types like locks are rejected with a helpful error.

Layer 2 — Worker Resource Runtime manages long-lived objects initialized once at worker startup. Resources are injected into tasks by name — no serialization needed, no connection per task.

Layer 3 — Resource Proxies handles objects that have capturable state: file handles, HTTP sessions, cloud clients. The interceptor extracts a recipe; the worker rebuilds the live object before the task runs.

Minimal end-to-end example

from flexiq import Queue, Inject
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

queue = Queue(db_path="tasks.db", interception="strict")


@queue.worker_resource("db")
def create_db():
    engine = create_engine("postgresql://localhost/myapp")
    return sessionmaker(engine)


@queue.task()
def process_order(order_id: int, db: Inject["db"]):
    session = db()
    try:
        order = session.get(Order, order_id)
        order.status = "processed"
        session.commit()
    finally:
        session.close()

The injected db is the sessionmaker returned by create_db(), not a session itself — call it (session = db()) to get a usable session, and close it when you're done.

Enqueue from anywhere in your application:

process_order.delay(42)
# The integer 42 passes through serialization normally.
# 'db' is injected by the worker — no session is ever put in the queue.

Start the worker:

flexiq worker --app myapp.tasks:queue
# [flexiq] Initialized 1 resource(s): db
# [flexiq] Worker started with 8 threads

Section overview