Managed Identities & Bring-Your-Own Connections¶

TaskQ ships DSN-first: the worker, client, admin UI, and migration helpers all accept a TASKQ_PG_DSN / TASKQ_REDIS_URL and construct asyncpg pools and redis.asyncio clients internally. That model breaks down when your deployment authenticates with rotating credentials: managed identities (Azure Entra ID, AWS IAM), dynamic secret managers (HashiCorp Vault), or short-lived OAuth tokens, all of which issue credentials that expire, so a DSN baked at process start goes stale mid-run.

This guide documents the connection hook points TaskQ exposes so you can either:

hand TaskQ a pre-constructed pool / connection / Redis client that you own and close yourself, or
hand TaskQ a zero-arg async factory that TaskQ invokes at the right point in its lifecycle to build that resource, letting you fetch a fresh credential at construction time.

If you run the stock taskq console script (including under the workgroup supervisor), you do not need a custom entrypoint to reach any of this: point TASKQ_PG_CREDENTIAL_PROVIDER / TASKQ_REDIS_CREDENTIAL_PROVIDER at your provider and TaskQ builds every factory for you; see From the CLI. Both are ordinary TaskQ settings, so they can equally live in a .env file alongside TASKQ_PG_DSN; the matching CLI flag, when given, wins.

For rotating-credential deployments, TaskQ provides a vendor-neutral credential provider abstraction (taskq.auth) with provider-specific implementations available as extras:

Extra	Module	Providers
`taskq[aad]`	`taskq.aad`	Microsoft Entra ID (Azure AD): PG + Redis
`taskq[aws]`	`taskq.aws`	AWS IAM RDS: PG
`taskq[vault]`	`taskq.vault`	HashiCorp Vault database secrets engine: PG
(none needed)	`taskq.auth`	Base interfaces + factory builders: implement your own provider

Why hook points (not just "pass a DSN")¶

Problem with DSN-only	What hooks give you
Token expires → new connections fail auth. Established sessions survive token expiry; it is pool growth / reconnects that get rejected.	Factory is called when TaskQ builds the pool; you fetch a fresh token then.
A password passed as a fixed string is reused for every connection the pool opens later.	The factory builders pass `password=` as an async callable, which asyncpg awaits once per physical connection - pool growth and idle-recycle replacements each authenticate with a freshly fetched credential.
Azure Redis requires a `CredentialProvider` returning `(username, token)` per reconnect.	You own the `redis.asyncio.Redis` client → pass a `CredentialProvider`.
You already run an app-wide pool (FastAPI lifespan) and want to share it.	Pass the pool directly; TaskQ will not close a caller-owned resource.
Migrations / `TaskQ.watch_reclaims()` open their own `asyncpg.connect(dsn)`.	`apply_pending_locked` and the client accept a `conn` / `conn_factory` so LISTEN/migrate work without a DSN.

Ownership rule (read this carefully)¶

Pre-constructed objects (pool=, redis_client=, notify_conn=, …) are caller-owned. TaskQ never closes them. You close them in your own lifespan/finally.

Factory-produced objects are TaskQ-owned. TaskQ closes them on teardown via its AsyncExitStack.

Three consequences:

A caller-owned notify_conn that drops leaves TaskQ nothing to rebuild through: the NOTIFY listener disables itself (logged as notify-listener-disabled) and the worker falls back to poll-based dispatch instead of crashing.
A caller-owned leader_conn with no leader_conn_factory and no pg_dsn_direct is a startup ValueError: a dropped leader connection could never be rebuilt, so TaskQ fails fast instead of silently never recovering leadership.
TaskQ-owned dedicated connections (DSN- or factory-built notify_conn / leader_conn) get TCP keepalive applied automatically.

Hook point inventory¶

Site	Pre-constructed	Factory	Notes
Worker: dispatcher pool	`WorkerConnections.dispatcher_pool`	`dispatcher_pool_factory`	`pg_dsn_direct` role
Worker: heartbeat pool	`WorkerConnections.heartbeat_pool`	`heartbeat_pool_factory`	`command_timeout=2s` is your responsibility when overriding
Worker: worker pool	`WorkerConnections.worker_pool`	`worker_pool_factory`	`pg_dsn_pooled` role
Worker: per-slot transaction pool	n/a (deliberately no `WorkerConnections` slot)	n/a (worker-internal)	Opened when a LOOP-scope `asyncpg.Connection` is registered and `max_concurrency > 1`; built by TaskQ on the direct DSN, or provider-backed via `pg_credential_provider` on `worker_main` / `worker_main_async`
Worker: notify conn	`WorkerConnections.notify_conn`	`notify_conn_factory`	LISTEN is issued by TaskQ; a dropped conn is rebuilt through the same factory
Worker: leader conn	`WorkerConnections.leader_conn`	`leader_conn_factory`	Advisory-lock conn
Worker: Redis	`WorkerConnections.redis_client`	`redis_client_factory`
Client: main pool	`TaskQ(pool=...)` (caller-owned)	`TaskQ(pool_factory=...)` or `TaskQ(dsn=..., pg_provider=...)`	TaskQ-owned; rotated on its `ReloadSchedule` (`reload_interval=` or the granted lease), on demand with `await tq.reload_credentials()`
Client: Redis	`TaskQ(redis_client=...)` ✓ existing	n/a
Client: `watch_reclaims` LISTEN conn	`TaskQ(listen_conn=...)`	`TaskQ(pg_conn_factory=...)`	Replaces the DSN-only LISTEN transport; `stream()` polls through the main pool and needs neither
Migrate: locked apply	`apply_pending_locked(conn=...)`	`apply_pending_locked(conn_factory=...)`	`list_applied` / `apply_pending` take an open conn only; no factory
Admin UI	`create_router(pg_pool=..., redis_client=...)` ✓ existing	n/a	Or `taskq ui serve --pg-credential-provider` / `--redis-credential-provider`
CLI: `taskq worker`	n/a	`--pg-credential-provider` / `--redis-credential-provider` (env: `TASKQ_PG_CREDENTIAL_PROVIDER` / `TASKQ_REDIS_CREDENTIAL_PROVIDER`)	Builds every worker role: all four pools (dispatcher, heartbeat, worker, and the conditional per-slot transaction pool), `notify_conn`, `leader_conn`, Redis
CLI: `taskq workgroup start`	n/a	same env vars	Children are `taskq worker` subprocesses and inherit the environment
CLI: `taskq migrate up` / `status`	n/a	`--pg-credential-provider` (env: `TASKQ_PG_CREDENTIAL_PROVIDER`)	One-shot connection through the provider

The credential provider abstraction¶

taskq.auth provides two async Protocols and reusable factory builders. Any provider implementing the Protocols gets all the factory builders for free: no third-party dependencies required.

Protocols¶

from taskq.auth import PgCredential, PgCredentialProvider, RedisCredential, RedisCredentialProvider


# Postgres: return a password (token) and optionally a fresh username
class PgCredentialProvider(Protocol):
    async def get_pg_credential(self) -> PgCredential: ...


# Redis: return (username, password)
class RedisCredentialProvider(Protocol):
    async def get_redis_credential(self) -> RedisCredential: ...

PgCredential carries a password (always required) and an optional username: token providers (AAD, AWS IAM) set only the password; dynamic username providers (Vault) set both. enrich_pg_dsn handles either case.

Factory builders¶

from taskq import make_pg_pool_factory, make_dedicated_conn_factory, make_redis_client_factory
# (also importable from taskq.auth)

# Any PgCredentialProvider → PoolFactory / ConnFactory
pool_factory = make_pg_pool_factory(dsn, provider, max_size=8, command_timeout=5.0)
conn_factory = make_dedicated_conn_factory(dsn, provider)

# Any RedisCredentialProvider → RedisFactory
redis_factory = make_redis_client_factory(url, provider)

The factories are zero-arg async callables matching the PoolFactory / ConnFactory / RedisFactory type aliases in taskq.connections. Each invocation fetches a fresh credential and calls asyncpg.create_pool / asyncpg.connect / redis.from_url. Redis reconnects re-fetch automatically via the redis-py CredentialProvider adapter.

Resilience kwargs are the factory's job. Clients TaskQ builds itself (from TASKQ_REDIS_URL) come from taskq._redis_client.build_redis_client, which sets health_check_interval=30, socket_keepalive=True, a 5s socket_timeout (pinned to redis-py 8.x's asyncio default — the bound the fail-closed wall stands on), and a 1-attempt exponential-backoff Retry (0.02s base, 0.1s cap — the retry buys the reconnect, never the outage) with retry_on_error=[ConnectionError, TimeoutError]. A factory-built or caller-supplied redis client skips that builder, so make_redis_client_factory callers should pass the same kwargs in its **client_kwargs (or pass the builder's output through and add credential_provider=):
from redis.asyncio.retry import Retry
from redis.backoff import ExponentialBackoff
from redis.exceptions import ConnectionError, TimeoutError

redis_factory = make_redis_client_factory(
    url,
    provider,
    health_check_interval=30,
    socket_keepalive=True,
    socket_timeout=5,
    retry=Retry(ExponentialBackoff(cap=0.1, base=0.02), retries=1),
    retry_on_error=[ConnectionError, TimeoutError],
)
Without them, Azure's ~10-minute idle-connection reap surfaces as raw ConnectionErrors: into JobHandle.progress_stream's / TaskQ.stream's pubsub consumers, and as rate-limit-redis-transient-retry / progress-publish-failure warning noise on the worker.

How the credential reaches asyncpg: the PG factories pass it as password= (always) and user= (when the credential carries one) keyword arguments. Keyword arguments take precedence over both DSN userinfo and DSN query parameters in asyncpg's resolver, so a stale credential baked into the DSN can never shadow the fresh one, and the token never appears in the DSN string.

enrich_pg_dsn(dsn, credential) (also exported from taskq top-level) is the string-helper variant for callers that need a self-contained DSN: the credential is written into the DSN userinfo (percent-encoded), replacing any existing userinfo password, and replacing the user only when credential.username is set (Vault dynamic creds). Never put the credential in the query string instead: asyncpg applies userinfo before query parameters, so a password= query param is silently ignored whenever the DSN already carries userinfo (enrich_pg_dsn drops stale user= / password= query params for the same reason).

sslmode: both the factories and enrich_pg_dsn add sslmode=require only when the DSN has no explicit sslmode; an explicit verify-ca / verify-full is preserved (never downgraded) and is recommended where your server presents a verifiable certificate. require encrypts the connection but does not** verify the server certificate.

A verifying sslmode needs an explicit sslrootcert. If you set verify-ca / verify-full, add a CA bundle to the DSN as well, e.g. &sslrootcert=/etc/ssl/certs/ca-certificates.crt, or set PGSSLROOTCERT in the environment. Neither asyncpg nor libpq falls back to the system trust store here: libpq documents that "one or more root certificates must be placed in the file ~/.postgresql/root.crt in the user's home directory", and asyncpg states that "default system root CA certificates won't be loaded when specifying a particular sslmode, following the same behavior in libpq". In a container whose home directory is empty, a verifying DSN with no sslrootcert therefore fails while asyncpg is parsing connection arguments, any TCP connection is attempted, with:

asyncpg.exceptions._base.ClientConfigurationError: root certificate file "/home/<user>/.postgresql/root.crt" does not exist or cannot be accessed

which does not resemble a certificate-verification failure and is easy to misdiagnose. TaskQ deliberately does not inject a CA path for you: choosing a trust root is a security decision, and the correct bundle is environment-specific (a private CA needs its own PEM).

From the CLI (no custom entrypoint)¶

taskq worker, taskq workgroup start, taskq ui serve and taskq migrate resolve a credential provider from a module:attr reference (the same syntax --actors uses):

export TASKQ_PG_CREDENTIAL_PROVIDER=myapp.auth:make_provider
export TASKQ_REDIS_CREDENTIAL_PROVIDER=myapp.auth:make_provider   # optional
taskq worker --actors myapp.actors:registry

# myapp/auth.py
from azure.identity.aio import DefaultAzureCredential
from taskq.aad import EntraIdProvider


def make_provider() -> EntraIdProvider:
    return EntraIdProvider(DefaultAzureCredential(), redis_username="<mi-object-id>")

The reference may point at any of:

a provider instance: myapp.auth:PROVIDER
a zero-arg factory returning one: myapp.auth:make_provider
the provider class, when its constructor takes no required arguments: myapp.auth:MyProvider

Equivalent CLI flags exist for one-off runs: --pg-credential-provider / --redis-credential-provider.

What the worker builds. Every Postgres role: the dispatcher, heartbeat and worker pools, the notify_conn LISTEN connection and the leader_conn advisory-lock connection), plus the Redis client is built through your provider, sized and timed out exactly as the DSN path sizes them (TASKQ_DISPATCHER_POOL_SIZE, TASKQ_HEARTBEAT_COMMAND_TIMEOUT, TASKQ_POOL_MAX_INACTIVE_LIFETIME, dispatcher_command_timeout on the dedicated connections). Because every role is factory-backed, SIGHUP and TASKQ_RELOAD_INTERVAL rotate all of them: a role left on the DSN fallback would be silently un-rotatable, since reload_credentials skips roles that have no factory.

The DSN is still read: it supplies host, port, database and (for username-stable providers) the user. The credential itself is passed to asyncpg as keyword arguments and never appears in the DSN string.

Workgroups. The supervisor spawns taskq worker subprocesses and does not scrub the environment, so exporting the env vars before taskq workgroup start configures every child. Send SIGHUP to the children (pkill -HUP -f 'taskq worker'), not to the supervisor.

Failures are fatal, never silent. A reference that does not import, names a missing attribute, resolves to something without an async get_pg_credential() / get_redis_credential(), or names a Redis provider with no TASKQ_REDIS_URL set, exits non-zero at startup. There is no fallback to the DSN password: a worker that quietly authenticated with a static credential would look healthy until the first reconnect after deploy.

Other commands.

# admin UI: pool and (with --migrate) the migration connection
taskq ui serve --pg-credential-provider myapp.auth:make_provider

# migrations: the one-shot connection
taskq migrate up --pg-credential-provider myapp.auth:make_provider

taskq ui serve rotates its admin pool the way the worker does: a token credential re-authenticates per physical connection through the password= callable described below, and a username-bearing pair (Vault) is rebuilt on SIGHUP and on the same TASKQ_RELOAD_INTERVAL / lease-derived cadence (see Token refresh for long-lived pools). Every admin route resolves the live pool per request, so a rotation never serves from a closed pool. The migration connection (--migrate) is one-shot and takes no schedule.

Embedding. The builder behind the worker option is public: use it when you have a custom entrypoint and want the same full wiring:

from taskq.auth import build_worker_connections

connections = build_worker_connections(settings, pg_provider=provider, redis_provider=provider)
worker_main(
    settings,
    actor_registry=ACTORS,
    connections=connections,
    pg_credential_provider=provider,  # the per-slot pool does not read WorkerConnections
)

The pg_credential_provider argument exists for the per-slot transaction pool (opened when a LOOP-scope asyncpg.Connection is registered and max_concurrency > 1): the role pools take their credentials from connections, the slot pool from this parameter, or from the direct DSN when omitted. See Worker: WorkerConnections.

Token refresh for long-lived pools¶

Token refresh is automatic - no external rotation schedule is required. The factory builders pass the token to asyncpg as a password= async callable (never baked into the DSN string), and asyncpg invokes and awaits it once per physical connection: the connections opened at pool creation, those opened later by pool growth, and the replacements opened after max_inactive_connection_lifetime recycles an idle connection. Every new connection therefore authenticates with a freshly fetched credential.

This matters because Postgres authenticates at connect time only. A credential resolved once and reused as a fixed string keeps working on already-open sessions while new connections fail auth - so a pool built that way is healthy at deploy and then, roughly one token lifetime later, degrades as idle connections are recycled. That was TaskQ's behaviour before this changed; it no longer is.

Two things per-connection refresh cannot do, for which the credential hot-reload below is still the answer:

A username-bearing pair. asyncpg resolves user= once per pool and accepts a callable only for password=, and a dynamic username is only valid with the password issued alongside it. Providers that issue a fresh username alongside each password (HashiCorp Vault dynamic database credentials) therefore pin one pair per pool: every physical connection authenticates with that pair's password, and the pair is replaced by the pool rebuild. The rebuild has to land before the issuer revokes the pair, and it does so by default: the cadence is derived from the granted lease when no interval is configured (below).
Forcing a full pool rebuild - e.g. to drop sessions opened under a revoked credential, or after a DSN/endpoint change.

The rebuild cadence: ReloadSchedule. Every pool and dedicated connection built by make_pg_pool_factory / make_dedicated_conn_factory records the credential it is issued on a taskq.auth.ReloadSchedule, and every consumer that rebuilds pools: the worker, taskq ui serve, TaskQ: reads its interval from that schedule:

TASKQ_RELOAD_INTERVAL (or TaskQ(reload_interval=...)) set: that interval, always.
Unset, and the provider reports a lease (PgCredential.lease_duration , VaultDynamicDbProvider reports the TTL Vault granted): half the shortest lease seen so far (taskq.auth.LEASE_RELOAD_FRACTION), so a rebuild that fails at T + TTL/2 still has a full half-life of retries before the pair is revoked. Logged at pool build as pg-lease-reload-derived (and by the worker as reload-interval-derived-from-lease). Shortest, not latest: Vault caps a lease at the issuing token's remaining TTL, so a lease can come back shorter than the last.
Unset, and the provider reports no lease: no timer. A username-bearing pool built this way warns once at build (pg-lease-pair-pinned-without-reload, naming TASKQ_RELOAD_INTERVAL) because reconnects will fail authentication once the lease expires; SIGHUP / deps.request_reload() / tq.reload_credentials() still rotate it. Token providers (username unset) never warn and never need a timer.

All four triggers run the same reload_credentials path:

Trigger	How	When to use
`TASKQ_RELOAD_INTERVAL` (seconds; unset = derived from the lease)	`TASKQ_RELOAD_INTERVAL=720 taskq worker --actors …`	Periodic reload with no external signal: the only option on Windows (no SIGHUP) and the hands-off option everywhere else. Set it to override the lease-derived cadence, or to force periodic rebuilds for a token provider (e.g. ~720 s for AWS IAM's 15-minute tokens).
SIGHUP	`pkill -HUP -f 'taskq worker'` (also `taskq ui serve`)	Unix on-demand rotation (cron, k8s CronJob, config-change hooks).
`deps.request_reload()`	programmatic, from an embedder holding `WorkerDeps`	In-process trigger (e.g. your own secrets-watch callback). Equivalent to SIGHUP.
`reload_credentials(deps, ...)`	direct async call	Lower-level (e.g. tests); returns `(reloaded, failed)`.

A reload only rotates factory-backed resources. On the console script that means the provider must be configured (TASKQ_PG_CREDENTIAL_PROVIDER, see From the CLI); on a pure DSN worker a reload reconnects notify_conn / leader_conn with the same static DSN credential and rotates nothing else. Confirm with the credentials-reloaded log line; its resources list names what actually rotated.

SIGHUP delivery patterns (the console script is taskq (there is no taskq-worker process name):

# cron / k8s CronJob / operator script
pkill -HUP -f 'taskq worker'

# systemd unit
ExecReload=/bin/kill -HUP $MAINPID

# container where `taskq worker` is PID 1 (exec-form entrypoint)
kill -HUP 1

What a reload does:

Every factory-backed pool, dedicated connection, and Redis client is rebuilt by re-invoking its factory (each factory call fetches a fresh credential). Each factory call is bounded by TASKQ_RELOAD_FACTORY_TIMEOUT (default 30 s) so a hung token endpoint cannot wedge the reload.
The swap is atomic: the old pool stops serving new acquisitions immediately, so new work starts on the new pool. DI-injected db: asyncpg.Pool actors also resolve the new pool (the LOOP-scope cache is refreshed after a successful worker-pool reload), and progress flushing follows the swap on its next tick.
The old pool is closed in the background with a bounded drain (drain_timeout, default 5 s): in-flight actors holding the old pool get that long to finish. On timeout the old pool is terminated; an actor that outlives the drain sees its next acquire() fail and the job retries, landing on the new pool.
A provider-backed per-slot transaction pool rotates with the rest, keeping its boot-time size (TASKQ_MAX_CONCURRENCY is boot-only). In-flight jobs get the drain timeout to finish; past it the old pool is terminated; job's transaction rolls back, the row stays running, and lock-lease expiry reclaims and retries it: a loud, retryable infrastructure failure, never a false success. For the drain window the worker briefly holds 2 × (max_concurrency + 1) slot-pool connections (old pool draining + new pool warm); budget for that peak when rotations can coincide across the fleet.
A SIGHUP arriving mid-reload (success or failure) is honored with exactly one follow-up reload; N signals during one reload coalesce into one follow-up, not N. Reloads are skipped while shutdown is in progress.
Job processing continues throughout: the dispatcher/consumer/ heartbeat loops are not stopped for a reload.

Each resource reloads independently: if one factory call fails (e.g. a transient credential-fetch error), that resource keeps its current pool/connection and everything else still reloads. Check the credentials-reloaded log line's failed field after a SIGHUP: a non-empty list means a partial reload; trigger another reload to retry the resources that didn't rotate.

leader_conn reload is indirect: closing it triggers the existing leader watchdog's reopen-and-re-acquire path (the same path a real connection drop takes), which also rebuilds the leader's other dedicated connections (the monitor and cron loops' connections) through the same credential source, so a single reload rotates every leader-owned connection, not just leader_conn itself. This happens within one heartbeat_interval tick, not instantly.

For AWS IAM RDS (15-minute tokens), rotate every ~12 minutes:

# hands-off (recommended)
export TASKQ_PG_CREDENTIAL_PROVIDER=myapp.auth:make_provider
TASKQ_RELOAD_INTERVAL=720 taskq worker --actors myapp.actors:registry

# or via cron / k8s CronJob
pkill -HUP -f 'taskq worker'

For Azure Redis, refresh is also automatic: redis-py calls the CredentialProvider on every reconnect, so a single factory-built client rotates tokens for free between reloads.

Caller-owned resources (passed as concrete pool= / redis_client= / notify_conn=) are not swapped by a reload; the caller owns their lifecycle. Only factory-backed resources are hot-reloaded.

reload_credentials can also be called programmatically (also re-exported from taskq.worker):

from taskq.worker.deps import reload_credentials

await reload_credentials(deps, drain_timeout=10.0, factory_timeout=30.0)

Provider extras¶

Azure Entra ID (AAD): `taskq[aad]`¶

pip install 'taskq-py[aad]'

The extra includes azure-identity and aiohttp (required by the azure.identity.aio async credentials).

from azure.identity.aio import DefaultAzureCredential
from taskq import (
    WorkerConnections,
    make_pg_pool_factory,
    make_dedicated_conn_factory,
    make_redis_client_factory,
)
from taskq.aad import EntraIdProvider

cred = DefaultAzureCredential()
provider = EntraIdProvider(cred, redis_username="<managed-identity-object-id>")

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,
    ),
    notify_conn_factory=make_dedicated_conn_factory(settings.pg_dsn_direct, provider),
    leader_conn_factory=make_dedicated_conn_factory(settings.pg_dsn_direct, provider),
    redis_client_factory=make_redis_client_factory(settings.redis_url, provider),
)

EntraIdProvider implements both Protocols: pass one instance to PG and Redis factories. For PG-only or Redis-only, use EntraIdPgProvider / EntraIdRedisProvider individually.

The providers accept either an async credential (azure.identity.aio, as above) or a sync azure.identity credential: sync credentials perform blocking HTTP, so their get_token is offloaded to a thread and never stalls the event loop. The credential you pass is caller-owned (close it in your lifespan). Pass credential=None (the default) and the provider lazily creates one DefaultAzureCredential and reuses it for its lifetime.

Scopes: https://ossrdbms-aad.database.windows.net/.default (PG), https://redis.azure.com/.default (Redis).

Prerequisites: enable Entra authentication on Azure DB for Postgres and Azure Cache for Redis; grant the managed identity the appropriate roles. The factories add sslmode=require when the DSN has no explicit sslmode (see sslmode under Factory builders); Azure requires TLS.

AWS IAM RDS: `taskq[aws]`¶

pip install 'taskq-py[aws]'

from taskq import WorkerConnections, make_pg_pool_factory, make_dedicated_conn_factory
from taskq.aws import RdsIamProvider

provider = RdsIamProvider(settings.pg_dsn_direct, region="us-east-1")

WorkerConnections(
    dispatcher_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_direct,
        provider,
        max_size=settings.dispatcher_pool_size,
    ),
    # ... heartbeat, worker, notify, leader similarly
)

AWS IAM RDS auth tokens are valid for 15 minutes. generate_db_auth_token itself is local SigV4 signing, but resolving the ambient AWS credential chain can block on STS/IMDS HTTPS refreshes, so the provider offloads the boto call to a thread rather than stalling the event loop, and builds its boto3.client('rds') once on first use (pass client= to supply your own). Pass region=None (the default) to let botocore fall back to the ambient client region. No reload schedule is needed for token freshness: every physical connection the pool opens signs a fresh token through the password= callable (see Token refresh for long-lived pools).

Prerequisites: enable IAM database authentication on the RDS instance; create an IAM-mapped DB user (GRANT rds_iam TO myuser); grant rds-db:connect to the IAM principal. The DSN's user must be the IAM-mapped DB user.

HashiCorp Vault: `taskq[vault]`¶

pip install 'taskq-py[vault]'

import hvac
from taskq import WorkerConnections, make_pg_pool_factory
from taskq.vault import VaultDynamicDbProvider

client = hvac.Client(url="https://vault.example", token="...")
provider = VaultDynamicDbProvider(client, role="taskq-readonly")

WorkerConnections(
    dispatcher_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_direct,
        provider,
        max_size=settings.dispatcher_pool_size,
    ),
    # ...
)

Vault's database secrets engine issues a fresh username + password on each generate_credentials call, with a configurable lease TTL. Unlike token providers, PgCredential.username is always set: the DSN's user is overridden, and the pair is only valid together, so the factory builders pin one lease per pool / dedicated connection: user= is the lease's username and every physical connection authenticates with that lease's password (a pool never burns a lease per connection). Rotation is the pool rebuild, and it is scheduled for you: the provider reports the TTL Vault granted as PgCredential.lease_duration, and with no TASKQ_RELOAD_INTERVAL set the worker, taskq ui serve and TaskQ rebuild each pool at half that TTL, before Vault revokes the previous user (see Token refresh for long-lived pools). Set TASKQ_RELOAD_INTERVAL to choose the cadence yourself. Each issued lease is logged as vault-lease-issued with its lease_id and lease_duration, and the derived cadence as pg-lease-reload-derived, so the rebuild schedule can be checked against the TTL Vault actually granted. A role whose secret carries no lease (lease_duration 0) derives nothing and warns at build. hvac is synchronous; the provider offloads generate_credentials to a thread via asyncio.to_thread.

Prerequisites: enable the database secrets engine; configure a connection and role pointing at your Postgres. The DSN's host/port/dbname must point at the Postgres Vault provisions creds for.

Other patterns (no extra needed)¶

These don't warrant a dedicated extra: implement a PgCredentialProvider or pass a pre-constructed pool directly.

GCP Cloud SQL IAM¶

Use the official google-cloud-sql-connector: it handles token refresh and mTLS automatically, so you don't need a credential provider. Build a pool factory on the connector's async API (create_async_connector+connect_asyncwith the"asyncpg"driver andenable_iam_auth=True), and pass it toasyncpg.create_poolvia the **connect=keyword**. All of these calls are already async, so the factory awaits them directly, with noasyncio.to_thread` offload:

from contextlib import asynccontextmanager

import asyncpg
from google.cloud.sql.connector import create_async_connector
from taskq import WorkerConnections


def make_cloudsql_pool_factory(connector, instance: str, user: str, db: str):
    """Zero-arg async pool factory backed by the Cloud SQL connector."""

    async def pg_pool_factory() -> asyncpg.Pool:
        async def getconn() -> asyncpg.Connection:
            return await connector.connect_async(
                instance,  # "project:region:instance"
                "asyncpg",  # driver: asyncpg, not pg8000
                user=user,  # IAM principal, e.g. "my-mi@project.iam"
                db=db,
                enable_iam_auth=True,
            )

        return await asyncpg.create_pool(connect=getconn)  # connect= keyword

    return pg_pool_factory


@asynccontextmanager
async def lifespan(app):
    connector = await create_async_connector()
    try:
        connections = WorkerConnections(
            dispatcher_pool_factory=make_cloudsql_pool_factory(
                connector, "project:region:instance", "my-mi@project.iam", "taskq"
            ),
            # ... heartbeat, worker, notify, leader similarly
        )
        # hand connections to worker_main / open_worker_deps
        yield
    finally:
        await connector.close_async()  # caller-owned: you close the connector

mTLS / client certificates¶

Pass an ssl.SSLContext via a factory: no credential provider needed:

import ssl

from taskq import WorkerConnections

sslctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
sslctx.load_cert_chain("client.crt", "client.key")


async def pg_pool_factory() -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn=settings.pg_dsn_direct,
        ssl=sslctx,
        min_size=1,
        max_size=4,
    )


WorkerConnections(dispatcher_pool_factory=pg_pool_factory)

Secrets-manager password rotation (AWS Secrets Manager, Doppler, etc.)¶

Implement a PgCredentialProvider that fetches the current password from your secrets manager on each call:

from taskq.auth import PgCredential, PgCredentialProvider


class SecretsManagerProvider:
    def __init__(self, client, secret_id: str) -> None:
        self._client = client
        self._secret_id = secret_id

    async def get_pg_credential(self) -> PgCredential:
        import asyncio, json

        def _fetch() -> str:
            resp = self._client.get_secret_value(SecretId=self._secret_id)
            return json.loads(resp["SecretString"])["password"]

        password = await asyncio.to_thread(_fetch)
        return PgCredential(password=password)

Custom OAuth / token endpoint¶

from taskq.auth import PgCredential, PgCredentialProvider
import httpx


class OAuthTokenProvider:
    def __init__(self, token_url: str, client_id: str, client_secret: str) -> None: ...

    async def get_pg_credential(self) -> PgCredential:
        async with httpx.AsyncClient() as client:
            resp = await client.post(self._token_url, data={...})
            token = resp.json()["access_token"]
        return PgCredential(password=token)

Worker: `WorkerConnections`¶

open_worker_deps and worker_main accept an optional connections: WorkerConnections dataclass. Any field left None falls back to the existing DSN construction, so the change is purely additive.

from taskq import WorkerConnections
from taskq.worker.run import worker_main

worker_main(
    settings,
    actor_registry=ACTORS,
    connections=WorkerConnections(
        dispatcher_pool_factory=my_factory,
        # fields left None → TaskQ builds them from DSNs as before.
    ),
)

Mixing a pre-constructed pool and a factory for the same role raises ValueError at startup: pick one.

The worker's per-slot transaction pool (opened when a LOOP-scope asyncpg.Connection is registered and max_concurrency > 1) has no WorkerConnections slot, by design: it is worker-internal infrastructure, built by TaskQ on the direct DSN. Embedders whose credentials live in their own pool factories must also pass pg_credential_provider to worker_main / worker_main_async: the same provider used to build connections, or the slot pool authenticates from the DSN (the worker warns slot_pool_own_credentials at boot when caller-supplied pools are in play with no provider). The taskq worker CLI passes the resolved --pg-credential-provider / TASKQ_PG_CREDENTIAL_PROVIDER automatically.

`worker_main_async`: when you must own the loop¶

worker_main drives its own asyncio.Runner. If your actors are closures over dependencies you have to build first. An asyncpg.Pool is loop-bound, so it must be created on the loop the worker will run on, so await worker_main_async from your own coroutine instead. Same parameters, same exit codes; worker_main is the thin sync wrapper over it.

import asyncio, asyncpg
from taskq.worker import worker_main_async


async def main() -> int:
    pool = await asyncpg.create_pool(dsn)  # this loop
    registry = build_actors(pool)  # actors close over it
    try:
        return await worker_main_async(settings, actor_registry=registry)
    finally:
        await pool.close()  # yours to close


raise SystemExit(asyncio.run(main()))

`build_worker_connections` with an explicit DSN¶

build_worker_connections(settings, pg_provider=…) reads its endpoints from TASKQ_PG_DSN. If your application already knows where TaskQ's tables live, pass that in rather than restating it in a second config system:

connections = build_worker_connections(
    worker_settings,
    pg_provider=provider,
    pg_dsn=my_settings.taskq_pg_dsn,  # or pg_dsn_direct=/pg_dsn_pooled= for a pgbouncer split
    redis_url=my_settings.taskq_redis_url,
)

Only the endpoint is overridden: every pool size and timeout still comes from WorkerSettings. That is the point: hand-building one make_pg_pool_factory and passing it to all three pool roles silently gives each role a full max_size pool (5 x pool_max + 3 per replica) instead of dispatcher_pool_size / heartbeat_pool_size / worker_pool_size.

You only need this when you have a custom entrypoint. Running the stock console script, set TASKQ_PG_CREDENTIAL_PROVIDER instead (From the CLI); with a custom entrypoint, taskq.auth.build_worker_connections(settings, pg_provider=…, redis_provider=…) builds the same complete set of factories in one call.

`PoolFactory` / `ConnFactory` / `RedisFactory` signatures¶

type PoolFactory = Callable[[], Awaitable[asyncpg.Pool]]
type ConnFactory = Callable[[], Awaitable[asyncpg.Connection]]
type RedisFactory = Callable[[], Awaitable[redis.asyncio.Redis]]

All three are zero-arg async callables: closures that capture whatever they need (DSN, sizing, credentials). Exported from taskq top-level.

Client: `TaskQ`¶

Rotating credentials on the client¶

TaskQ takes a pool_factory=: the same PoolFactory the worker takes per role. TaskQ invokes it at open() and owns the result, and await tq.reload_credentials() re-invokes it to swap the pool in place: the backend behind enqueue/get/list/cancel, the tq.actors client and the stream() poll all move to the new pool, and the old one is closed with a bounded drain. Nothing needs to reach into the client's internals, and no restart is required when a token expires.

A username-bearing pair is rotated automatically: the client runs the same rebuild loop as the worker for the life of the client, at reload_interval= when given and otherwise at half the lease TTL the factory was granted (a token provider's pool has no lease and is never rebuilt by itself). A rebuild that fails leaves the live pool serving and is retried on the next tick; reload_credentials() remains the on-demand path and is serialized with the scheduled one.

from taskq import TaskQ
from taskq.aad import EntraIdProvider

provider = EntraIdProvider()

# Sugar: dsn + provider. Exactly make_pg_pool_factory(dsn, provider,
# min_size=min_pool_size, max_size=max_pool_size) under the hood.
tq = TaskQ(dsn=dsn, pg_provider=provider, max_pool_size=5)
await tq.open()

# ...anywhere: a scheduled task, a signal handler, your issuer's callback.
await tq.reload_credentials()

Use pool_factory= directly when you need the factory's other hooks (init=, server_settings=, command_timeout=):

from taskq import make_pg_pool_factory

tq = TaskQ(pool_factory=make_pg_pool_factory(dsn, provider, max_size=5, init=register_vector))

dsn=, pool= and pool_factory= are mutually exclusive, and pg_provider= requires dsn=. Ownership follows the rule above: a pool TaskQ built (from a DSN or a factory) is closed by tq.close(); one you passed as pool= never is, and reload_credentials() refuses to rotate it, because rotating means closing.

reload_credentials() is not needed for ordinary token refresh: make_pg_pool_factory passes password= to asyncpg as a callable, so every new physical connection already authenticates with a freshly fetched credential. Reload is how you drop sessions opened under a revoked credential, and the way a username-bearing pair such as a Vault lease rotates (asyncpg resolves user= once per pool, so the pool stays on the pair it was built with), scheduled for you as described above.

LISTEN transport¶

TaskQ accepts pool= and redis_client= (caller-owned). Two additions close the remaining DSN-only gaps for the LISTEN/NOTIFY transport in watch_reclaims():

from taskq import make_dedicated_conn_factory

tq = TaskQ(
    pool=app_state.pg_pool,  # caller-owned
    redis_client=app_state.redis,  # caller-owned
    # LISTEN transport for tq.watch_reclaims() without a DSN:
    pg_conn_factory=make_dedicated_conn_factory(settings.pg_dsn_direct, provider),  # OR
    listen_conn=app_state.listen_conn,  # pre-constructed, caller-owned
)

tq.stream() needs neither: on Postgres alone it re-reads the job row through the main pool every min(poll_timeout, 0.5) seconds (nothing announces a job's transitions over NOTIFY), so it works in pool-only mode and holds no connection per stream. With redis_client= it subscribes to the job's progress channel instead.

`ensure_sslmode_require`¶

Every factory in taskq.auth applies this before connecting. Call it yourself on the DSN paths they do not cover: a raw asyncpg.connect, a migration connection, a DSN handed to another library:

from taskq import ensure_sslmode_require

conn = await asyncpg.connect(ensure_sslmode_require(dsn), password=token)

It adds sslmode=require only when no sslmode is set. An explicit mode is never overridden: verify-ca / verify-full are not downgraded (require skips certificate verification, which would expose the very token you are injecting), and sslmode=disable stays disabled, which is how a test container or a Unix-socket deployment opts out.

Because an explicit verify-ca / verify-full is passed straight through to asyncpg, it must carry its own sslrootcert (or PGSSLROOTCERT); see sslmode under Factory builders above. This helper only ever fills in a missing sslmode; it does not, and should not, supply a CA bundle.

Migrate¶

from taskq.migrate import apply_pending_locked

await apply_pending_locked(conn_factory=lambda: build_conn(token), schema="taskq")

conn (caller-owned) and conn_factory (TaskQ-owned) are mutually exclusive; either replaces the dsn parameter.

From the CLI, taskq migrate up --pg-credential-provider myapp.auth:make_provider (also migrate status, and ui serve --migrate) builds that conn_factory for you.

FastAPI lifespan example¶

from contextlib import asynccontextmanager
from azure.identity.aio import DefaultAzureCredential
from taskq import TaskQ, make_pg_pool_factory, make_redis_client_factory
from taskq.aad import EntraIdProvider


@asynccontextmanager
async def lifespan(app):
    # The credential is caller-owned: async with (or aclose() in a
    # finally) so its aiohttp session is closed on shutdown.
    async with DefaultAzureCredential() as cred:
        provider = EntraIdProvider(cred)
        # Factory-build a caller-owned pool (fresh AAD token at construction)
        # and a caller-owned Redis client (auto-rotating tokens on reconnect).
        pool_factory = make_pg_pool_factory(settings.pg_dsn_direct, provider, max_size=5)
        pg_pool = await pool_factory()
        redis_factory = make_redis_client_factory(settings.redis_url, provider)
        redis_client = await redis_factory()
        try:
            app.state.tq = TaskQ(
                pool=pg_pool,  # caller-owned
                redis_client=redis_client,  # caller-owned
            )
            await app.state.tq.open()
            yield
        finally:
            await app.state.tq.close()
            await pg_pool.close()
            await redis_client.aclose()

See examples/fastapi_app/aad.py for a runnable end-to-end scaffold.