Connections (BYO / Factories)

Connection hook points — WorkerConnections per-role overrides and the PoolFactory / ConnFactory / RedisFactory type aliases. See the Managed Identities & BYO Connections guide for the ownership model and deployment patterns.

connections

Connection hook points — bring-your-own resources or factories.

TaskQ constructs its asyncpg pools, dedicated connections, and Redis client internally from DSN strings by default. This module lets you replace any of those with either:

1. a pre-constructed, caller-owned resource (TaskQ uses it but never closes it — you close it in your own lifespan), or
2. a zero-arg async factory that TaskQ invokes at the right point in its lifecycle and closes the result of on teardown (TaskQ-owned).

Fields left None fall back to the existing DSN construction, so the hook points are purely additive.

See the managed-identities deployment guide (docs/guides/managed-identities.md); :mod:taskq.auth provides vendor-neutral credential providers and factory builders, with provider-specific implementations in :mod:taskq.aad, :mod:taskq.aws, and :mod:taskq.vault.

Ownership rule

• Pre-constructed objects are caller-owned. TaskQ never closes them — close them in your own finally / lifespan.
• Factory-produced objects are TaskQ-owned. TaskQ closes them on teardown via its :class:~contextlib.AsyncExitStack.

Passing both a concrete resource and a factory for the same role is a configuration error (caught in :meth:WorkerConnections.__post_init__).

__all__ module-attribute

__all__ = [
    "ConnFactory",
    "PoolFactory",
    "RedisFactory",
    "WorkerConnections",
]

PoolFactory

PoolFactory = Callable[[], Awaitable[Pool]]

ConnFactory

ConnFactory = Callable[[], Awaitable[Connection]]

RedisFactory

RedisFactory = Callable[[], Awaitable[Redis]]

WorkerConnections dataclass

WorkerConnections(
    dispatcher_pool: Pool | None = None,
    dispatcher_pool_factory: PoolFactory | None = None,
    heartbeat_pool: Pool | None = None,
    heartbeat_pool_factory: PoolFactory | None = None,
    worker_pool: Pool | None = None,
    worker_pool_factory: PoolFactory | None = None,
    notify_conn: Connection | None = None,
    notify_conn_factory: ConnFactory | None = None,
    leader_conn: Connection | None = None,
    leader_conn_factory: ConnFactory | None = None,
    redis_client: Redis | None = None,
    redis_client_factory: RedisFactory | None = None,
)

Per-role connection overrides for the worker.

Each role has a <role> (pre-constructed, caller-owned) and a <role>_factory (zero-arg async factory, TaskQ-owned) slot. Leave both None for DSN-based construction (the default).

Example — AAD-managed-identity worker::

from azure.identity.aio import DefaultAzureCredential
from taskq.aad import EntraIdProvider
from taskq.auth import make_pg_pool_factory
from taskq.connections import WorkerConnections

cred = DefaultAzureCredential()
provider = EntraIdProvider(cred)

connections = WorkerConnections(
    dispatcher_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_direct, provider, max_size=settings.dispatcher_pool_size,
    ),
    heartbeat_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_direct, provider,
        max_size=settings.heartbeat_pool_size,
        command_timeout=settings.heartbeat_command_timeout,
    ),
    worker_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_pooled, provider, max_size=settings.worker_pool_size,
    ),
)

Example — share an app-wide pool (caller-owned)::

connections = WorkerConnections(worker_pool=app_state.pg_pool)

dispatcher_pool class-attribute instance-attribute

dispatcher_pool: Pool | None = None

Dispatcher pool (pg_dsn_direct role). Caller-owned if set.

dispatcher_pool_factory class-attribute instance-attribute

dispatcher_pool_factory: PoolFactory | None = None

Factory for the dispatcher pool. TaskQ-owned.

heartbeat_pool class-attribute instance-attribute

heartbeat_pool: Pool | None = None

Heartbeat pool (pg_dsn_direct, heartbeat_command_timeout). Caller-owned.

heartbeat_pool_factory class-attribute instance-attribute

heartbeat_pool_factory: PoolFactory | None = None

Factory for the heartbeat pool. TaskQ-owned. command_timeout is your responsibility when overriding — set it on create_pool.

worker_pool class-attribute instance-attribute

worker_pool: Pool | None = None

Worker pool (pg_dsn_pooled role). Caller-owned.

worker_pool_factory class-attribute instance-attribute

worker_pool_factory: PoolFactory | None = None

Factory for the worker pool. TaskQ-owned.

notify_conn class-attribute instance-attribute

notify_conn: Connection | None = None

Dedicated LISTEN connection. Caller-owned. TaskQ still issues LISTEN.

notify_conn_factory class-attribute instance-attribute

notify_conn_factory: ConnFactory | None = None

Factory for the LISTEN connection. TaskQ-owned.

leader_conn class-attribute instance-attribute

leader_conn: Connection | None = None

Dedicated advisory-lock connection. Caller-owned.

leader_conn_factory class-attribute instance-attribute

leader_conn_factory: ConnFactory | None = None

Factory for the advisory-lock connection. TaskQ-owned.

redis_client class-attribute instance-attribute

redis_client: Redis | None = None

Redis client for progress fanout / rate limiting. Caller-owned.

redis_client_factory class-attribute instance-attribute

redis_client_factory: RedisFactory | None = None

Factory for the Redis client. TaskQ-owned.

__post_init__

__post_init__() -> None

Reject concrete + factory for the same role (configuration error).

Source code in src/taskq/connections.py

def __post_init__(self) -> None:
    """Reject concrete + factory for the same role (configuration error)."""
    for concrete, factory in (
        ("dispatcher_pool", "dispatcher_pool_factory"),
        ("heartbeat_pool", "heartbeat_pool_factory"),
        ("worker_pool", "worker_pool_factory"),
        ("notify_conn", "notify_conn_factory"),
        ("leader_conn", "leader_conn_factory"),
        ("redis_client", "redis_client_factory"),
    ):
        if getattr(self, concrete) is not None and getattr(self, factory) is not None:
            raise ValueError(
                f"WorkerConnections: provide either {concrete!r} or "
                f"{factory!r}, not both (role would be ambiguous)."
            )

has_any

has_any() -> bool

True if any override (concrete or factory) is set.

Source code in src/taskq/connections.py

def has_any(self) -> bool:
    """True if any override (concrete or factory) is set."""
    return any(
        getattr(self, name) is not None
        for name in (
            "dispatcher_pool",
            "dispatcher_pool_factory",
            "heartbeat_pool",
            "heartbeat_pool_factory",
            "worker_pool",
            "worker_pool_factory",
            "notify_conn",
            "notify_conn_factory",
            "leader_conn",
            "leader_conn_factory",
            "redis_client",
            "redis_client_factory",
        )
    )