Resources & Locking

Methods for the worker resource system and distributed locking.

Resource system

@queue.worker_resource()

@queue.worker_resource(
    name: str,
    depends_on: list[str] | None = None,
    teardown: Callable | None = None,
    health_check: Callable | None = None,
    health_check_interval: float = 0.0,
    max_recreation_attempts: int = 3,
    scope: ResourceScope | str = ResourceScope.WORKER,
    pool_size: int | None = None,
    pool_min: int = 0,
    acquire_timeout: float = 10.0,
    max_lifetime: float = 3600.0,
    idle_timeout: float = 300.0,
    reloadable: bool = False,
    frozen: bool = False,
) -> Callable

Decorator to register a resource factory initialized at worker startup.

queue.register_resource()

queue.register_resource(definition: ResourceDefinition) -> None

Programmatically register a ResourceDefinition. Equivalent to @queue.worker_resource() but accepts a pre-built definition object.

queue.load_resources()

queue.load_resources(toml_path: str) -> None

Load resource definitions from a TOML file. Must be called before run_worker(). See TOML configuration.

queue.health_check()

queue.health_check(name: str) -> bool

Run a resource's health check immediately. Returns True if healthy, False otherwise or if the runtime is not initialized.

queue.reload_resources()

queue.reload_resources(names: list[str] | None = None) -> dict[str, bool]

Hot-reload reloadable worker resources — the programmatic equivalent of SIGHUP. Tears down and re-creates each target in dependency order. None reloads every resource registered with reloadable=True. Returns a name-to-success mapping; empty when no worker resource runtime is active in this process. Async twin: areload_resources().

queue.resource_status()

queue.resource_status() -> list[dict]

Return per-resource status. Each entry contains: name, scope, health, init_duration_ms, recreations, depends_on. Pooled entries also include pool stats.

queue.register_type()

queue.register_type(
    python_type: type,
    strategy: str,
    *,
    resource: str | None = None,
    message: str | None = None,
    converter: Callable | None = None,
    type_key: str | None = None,
    proxy_handler: str | None = None,
) -> None

Register a custom type with the interception system. Requires interception="strict" or "lenient".

queue.analyze_arguments()

queue.analyze_arguments(args: tuple = (), kwargs: dict | None = None) -> InterceptionReport

Dry-run the argument interceptor without enqueuing — the report describes the strategy chosen for each argument. Empty when the queue was created with interception="off". Per-task variant: my_task.analyze(...).

queue.interception_stats()

queue.interception_stats() -> dict

Return interception metrics: total call count, per-strategy counts, average duration in ms, max depth reached. Returns an empty dict if interception is disabled.

queue.proxy_stats()

queue.proxy_stats() -> list[dict]

Return per-handler proxy metrics: handler name, deconstruction count, reconstruction count, error count, average reconstruction time in ms.

Distributed locking

queue.lock()

queue.lock(
    name: str,
    ttl: int = 30,
    auto_extend: bool = True,
    owner_id: str | None = None,
    timeout: float | None = None,
    retry_interval: float = 0.1,
) -> contextlib.AbstractContextManager

Acquire a distributed lock. Use as a context manager:

with queue.lock("my-resource", ttl=60):
    # exclusive section
    ...

Raises LockNotAcquired if acquisition fails (when timeout is None or expires).

queue.alock()

queue.alock(
    name: str,
    ttl: float = 30.0,
    auto_extend: bool = True,
    owner_id: str | None = None,
    timeout: float | None = None,
    retry_interval: float = 0.1,
) -> AsyncDistributedLock

Async context manager version of lock(). Returns an AsyncDistributedLock directly — use async with, not await:

async with queue.alock("my-resource"):
    ...