Dependency Injection

worker_resource decorator, scopes (worker/thread/task/pooled), dependencies, teardown, health checks.

Worker resources are long-lived objects initialized once at worker startup and injected into tasks by name. No serialization is involved — they live entirely in the worker process and are never put in the queue.

Why not just a module-level global?

Celery has no first-class DI, so you typically open a connection at import time (db = create_engine(...)) and share that global. Worker resources replace that pattern: flexiq creates one instance per worker, scopes it (worker / thread / task / pooled), injects it by name, and tears it down on shutdown — no import-time side effects, and clean per-scope lifecycles.

Declaring resources

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

The factory runs once when the worker starts. The return value is the resource instance.

The factory can be async:

python

@queue.worker_resource("redis")
async def create_redis():
    import redis.asyncio as aioredis
    return await aioredis.from_url("redis://localhost")

FlexiQ runs the async factory on the worker's event loop before accepting tasks.

`worker_resource()` parameters

Parameter	Default	Description
`name`	required	Resource name used in `inject=["name"]` and `Inject["name"]`.
`depends_on`	`[]`	Names of resources this factory receives as arguments.
`teardown`	`None`	Callable invoked with the instance on graceful shutdown.
`health_check`	`None`	Callable invoked periodically; returns truthy if healthy.
`health_check_interval`	`0.0`	Seconds between health checks. `0` disables checking.
`max_recreation_attempts`	`3`	Max times to recreate after consecutive health failures.
`scope`	`"worker"`	Lifetime scope — see Resource scopes below.
`pool_size`	`None`	Pooled scope: max concurrent instances (default: 4).
`pool_min`	`0`	Pooled scope: pre-warmed instances at startup.
`acquire_timeout`	`10.0`	Pooled scope: seconds to wait for an available instance.
`max_lifetime`	`3600.0`	Pooled scope: max seconds an instance can live.
`idle_timeout`	`300.0`	Pooled scope: max idle seconds before eviction.
`reloadable`	`False`	Allow hot reload via SIGHUP or CLI.
`frozen`	`False`	Wrap instance in a read-only wrapper (see Configuration).

Injecting resources into tasks

python

@queue.task(inject=["db"])
def process_order(order_id: int, db):
    session = db()
    order = session.get(Order, order_id)
    ...

python

from flexiq import Inject

@queue.task()
def process_order(order_id: int, db: Inject["db"]):
    session = db()
    order = session.get(Order, order_id)
    ...

Both syntaxes are equivalent. Inject["name"] is a type annotation — it works with any type checker and makes the dependency explicit in the function signature. The worker reads the annotation at task registration time and injects the resource automatically.

If a caller explicitly passes a db kwarg to .delay(), that value wins over injection.

Resource scopes

Scope	Lifetime	Use case
`"worker"` (default)	Entire worker process	Database connection pools, shared caches
`"thread"`	One instance per worker thread, created lazily	Thread-unsafe objects that must not be shared
`"task"`	Fresh instance per task, torn down after	Stateful per-task objects
`"pooled"`	Checked out of a bounded pool per task, returned after	Short-lived connections with limited concurrency

Scope names and their meanings are shared across SDKs. There is no "request" scope (a fresh instance per resolve) here: resources arrive by injection, resolved once per task, so there is no second resolve to build for.

python

# Pooled scope: each task checks out a session from a pool of up to 10
@queue.worker_resource("db_session", scope="pooled", pool_size=10, depends_on=["db"])
def create_session(db):
    return db()  # db must be a worker-scoped resource

# Thread scope: one cache per worker thread
@queue.worker_resource("local_cache", scope="thread")
def create_cache():
    return {}

# Task scope: a fresh instance for every task, torn down right after
@queue.worker_resource("audit_context", scope="task")
def create_audit_context():
    return AuditContext(started_at=time.time())

Pool configuration parameters (pool_size, pool_min, acquire_timeout, max_lifetime, idle_timeout) only apply to pooled resources — passing pool_size or pool_min with any other scope raises. See Configuration for details.

Breaking rename. "task" used to mean checked out of a pool and "request" meant fresh per task; they are now "pooled" and "task", the names the other SDKs already used. scope="request" raises. scope="task" still resolves — but now builds per task instead of pooling, so move pooled resources to scope="pooled". Pool settings on a non-pooled scope raise, which catches the common case.

Dependencies

Resources can declare other resources they depend on. FlexiQ resolves the dependency graph and initializes in topological order, injecting dependencies as keyword arguments to the factory:

python

@queue.worker_resource("config")
def load_config():
    return Config.from_env()


@queue.worker_resource("db", depends_on=["config"])
def create_db(config):
    # Return a sessionmaker so the injected `db` is callable: `session = db()`.
    # In production, wire a teardown to dispose the engine (see Teardown below).
    return sessionmaker(create_engine(config.db_url, pool_size=10))


@queue.worker_resource("cache", depends_on=["config"])
def create_cache(config):
    return Redis.from_url(config.redis_url)

On shutdown, resources are torn down in reverse initialization order — cache and db before config.

Cycles are detected eagerly at registration time and raise CircularDependencyError.

Teardown

Supply a teardown callable to clean up the resource on graceful shutdown. Build the engine inside the factory — not at module scope — so it's created within the resource lifecycle, then dispose it in teardown to release the connection pool:

python

@queue.worker_resource(
    "db",
    # The injected `db` is the sessionmaker; `db.kw["bind"]` is its engine.
    teardown=lambda db: db.kw["bind"].dispose(),
)
def create_db():
    engine = create_engine("postgresql://localhost/myapp")
    return sessionmaker(engine)

Or use register_resource() for the programmatic API:

python

from flexiq.resources.definition import ResourceDefinition

queue.register_resource(ResourceDefinition(
    name="db",
    factory=create_db,
    teardown=close_db,
    depends_on=["config"],
))

Teardown callables can be async — FlexiQ awaits them if they return a coroutine.

Health checking

Resources can declare a health check function that runs on a background thread. If the check returns falsy, the worker attempts to recreate the resource:

python

def check_db(db):
    session = db()               # `db` is the sessionmaker returned by the factory
    try:
        session.execute(text("SELECT 1"))
        return True
    finally:
        session.close()


@queue.worker_resource(
    "db",
    health_check=check_db,
    health_check_interval=30.0,   # check every 30 seconds
    max_recreation_attempts=3,    # mark permanently unhealthy after 3 failures
)
def create_db():
    return sessionmaker(create_engine("postgresql://localhost/myapp"))

The health checker runs in a single daemon thread. Each resource with a non-zero health_check_interval is checked independently on its own schedule.

Run a health check manually from application code:

python

is_healthy = queue.health_check("db")

If a resource fails all recreation attempts, it is marked permanently unhealthy. Subsequent tasks that depend on it raise ResourceUnavailableError.

Resource status

python

status = queue.resource_status()
# [
#   {
#     "name": "config",
#     "scope": "worker",
#     "health": "healthy",
#     "init_duration_ms": 12.4,
#     "recreations": 0,
#     "depends_on": [],
#   },
#   {
#     "name": "db",
#     "scope": "worker",
#     "health": "healthy",
#     "init_duration_ms": 45.2,
#     "recreations": 0,
#     "depends_on": ["config"],
#   },
# ]

Task-scoped resources include a "pool" key with pool statistics. See Observability for details.

Full example

python

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

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


@queue.worker_resource("config")
def load_config():
    return Config.from_env()


def check_db(engine):
    with engine.connect() as conn:
        conn.execute(text("SELECT 1"))
    return True


@queue.worker_resource(
    "db",
    depends_on=["config"],
    teardown=lambda engine: engine.dispose(),
    health_check=check_db,
    health_check_interval=60.0,
)
def create_db(config):
    return create_engine(config.database_url, pool_size=10)


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