Web

Admin UI and progress streaming routers.

taskq.web is an empty module (docstring only) — it renders no members. The directives below target the concrete submodules instead. See also the Admin UI guide and Progress & Streaming guide for task-oriented walkthroughs.

Progress SSE router

progress

FastAPI router: SSE progress bridge and poll-state endpoint.

Bridges Redis pub/sub progress events to Server-Sent Events for browsers and API clients. Mount at prefix="/jobs" to produce the canonical URLs:

GET /jobs/api/job/{job_id}/progress/stream  — SSE stream
GET /jobs/api/job/{job_id}/state            — poll-state (JSON)

Importing this module requires the taskq[fastapi] optional extra (which includes sse-starlette).

Design notes

• Subscribe-before-query: the Redis channel is subscribed BEFORE the PG snapshot read to eliminate the race window where an event published between the PG read and the subscribe would be silently lost.
• The PG connection is released immediately after the initial row fetch; no PG connection is held during streaming.
• The Redis subscribe and PG query happen in the handler body (before EventSourceResponse is created) so that 404/503 errors produce proper HTTP status codes rather than appearing inside an already-started SSE stream.
• Keepalive comments are emitted every sse_heartbeat_interval seconds via a get_message(timeout=...) polling loop (avoids blocking listen() which has no per-message timeout support).
• On client disconnect, try/finally in the generator calls pubsub.unsubscribe() and pubsub.aclose() to prevent stale Redis subscriptions.

logger module-attribute

logger = structlog.get_logger('taskq.web.progress')

create_router

create_router(
    pg_pool: Pool,
    redis_client: Any,
    *,
    schema: str = "taskq",
    auth_dependency: Callable[..., Any] | None = None,
    sse_heartbeat_interval: timedelta = timedelta(
        seconds=15
    ),
    max_sse_connections: int | None = None,
) -> APIRouter

Return a FastAPI APIRouter exposing the SSE progress bridge.

Mount at prefix="/jobs" to produce the canonical paths::

GET /jobs/api/job/{job_id}/progress/stream
GET /jobs/api/job/{job_id}/state

Parameters

pg_pool: asyncpg connection pool for snapshot reads and poll-state queries. redis_client: redis.asyncio.Redis instance, or None when Redis is not configured. If None, the SSE endpoint returns HTTP 503. schema: PostgreSQL schema name (default "taskq"). auth_dependency: Optional FastAPI dependency callable; if provided it is injected via Depends() on all routes (same pattern as taskq.web.admin.create_router). sse_heartbeat_interval: Cadence for ': keepalive' SSE comments (default 15 s). max_sse_connections: Maximum concurrent progress streams this process will serve; further connections get HTTP 429. Defaults to TASKQ_PROGRESS_MAX_SSE_CONNECTIONS (50). Each stream holds a Redis pubsub subscription and an asyncio task for as long as the client stays connected, so an uncapped endpoint lets any principal who can reach the route exhaust Redis connections, event-loop tasks and file descriptors on the app hosting the pipeline. The admin /sse/{topic} endpoint has had such a cap; this one did not.

Source code in src/taskq/web/progress.py

def create_router(
    pg_pool: asyncpg.Pool,
    redis_client: Any,  # redis.asyncio.Redis | None — typed Any at erasure boundary; redis is an optional dep
    *,
    schema: str = "taskq",
    auth_dependency: Callable[..., Any] | None = None,
    sse_heartbeat_interval: timedelta = timedelta(seconds=15),
    max_sse_connections: int | None = None,
) -> APIRouter:
    """Return a FastAPI ``APIRouter`` exposing the SSE progress bridge.

    Mount at ``prefix="/jobs"`` to produce the canonical paths::

        GET /jobs/api/job/{job_id}/progress/stream
        GET /jobs/api/job/{job_id}/state

    Parameters
    ----------
    pg_pool:
        asyncpg connection pool for snapshot reads and poll-state queries.
    redis_client:
        ``redis.asyncio.Redis`` instance, or ``None`` when Redis is not
        configured.  If ``None``, the SSE endpoint returns HTTP 503.
    schema:
        PostgreSQL schema name (default ``"taskq"``).
    auth_dependency:
        Optional FastAPI dependency callable; if provided it is injected via
        ``Depends()`` on all routes (same pattern as
        ``taskq.web.admin.create_router``).
    sse_heartbeat_interval:
        Cadence for ``': keepalive'`` SSE comments (default 15 s).
    max_sse_connections:
        Maximum concurrent progress streams this process will serve; further
        connections get HTTP 429. Defaults to
        ``TASKQ_PROGRESS_MAX_SSE_CONNECTIONS`` (50). Each stream holds a Redis
        pubsub subscription and an asyncio task for as long as the client stays
        connected, so an uncapped endpoint lets any principal who can reach the
        route exhaust Redis connections, event-loop tasks and file descriptors
        on the app hosting the pipeline. The admin ``/sse/{topic}`` endpoint has
        had such a cap; this one did not.
    """
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema identifier: {schema!r}")

    router_kwargs: dict[str, Any] = {"tags": ["progress"]}
    if auth_dependency is not None:
        router_kwargs["dependencies"] = [Depends(auth_dependency)]

    router = APIRouter(**router_kwargs)

    _schema = schema
    _redis_client = redis_client
    _pg_pool = pg_pool
    _heartbeat_secs = sse_heartbeat_interval.total_seconds()
    if max_sse_connections is None:
        max_sse_connections = TaskQSettings.load().progress_max_sse_connections
    _max_sse = max_sse_connections
    _progress_sql = _PROGRESS_SQL.format(schema=_schema)

    # ----------------------------------------------------------------
    # SSE endpoint
    # ----------------------------------------------------------------

    @router.get(
        "/api/job/{job_id}/progress/stream",
        response_class=EventSourceResponse,
    )
    async def progress_stream(  # pyright: ignore[reportUnusedFunction]  # Why: registered via FastAPI decorator; pyright cannot see the route registration.
        job_id: UUID,
        request: Request,
        last_event_id: int | None = None,
    ) -> Response:
        """Stream progress events for a job via SSE.

        On initial connection (no ``last_event_id`` / ``Last-Event-ID``
        header): emits the current PG snapshot, then streams Redis events.

        On reconnect (``last_event_id`` present): subscribes Redis FIRST,
        emits one catch-up event from PG if ``progress_seq > last_event_id``,
        then resumes streaming.

        HTTP 404 — job not found.
        HTTP 503 — Redis not configured or unavailable.
        """
        if _redis_client is None:
            return JSONResponse(  # pyright: ignore[reportReturnType]  # Why: FastAPI accepts any Response subclass here; JSONResponse is returned for the 503 before SSE upgrade.
                status_code=503,
                content=_REDIS_503_BODY,
                headers={"Retry-After": "2"},
            )

        # Why here: after the cheap 503 guard (so an unconfigured Redis does
        # not consume a slot) and before any pubsub subscription is created,
        # so a rejected connection allocates nothing.
        sse_slot_semaphore = await acquire_sse_slot("progress-stream", _max_sse)
        # The slot is held for the LIFE of the stream, so ownership transfers
        # to the generator on the success path only. Every early exit below
        # (503 subscribe failure, PG error, 404 job not found) has to give it
        # back -- otherwise a run of requests for missing jobs would exhaust
        # the cap without a single stream ever opening. Idempotent so the
        # generator's own release can never double-release.
        _slot_released = False

        def _release_slot() -> None:
            nonlocal _slot_released
            if not _slot_released:
                _slot_released = True
                sse_slot_semaphore.release()

        try:
            return await _serve_progress_stream(
                job_id=job_id,
                request=request,
                last_event_id=last_event_id,
                sse_slot_semaphore=sse_slot_semaphore,
                release_slot=_release_slot,
            )
        except BaseException:
            _release_slot()
            raise

    async def _serve_progress_stream(  # pyright: ignore[reportUnusedFunction]  # Why: called by progress_stream above; not a route.
        *,
        job_id: UUID,
        request: Request,
        last_event_id: int | None,
        sse_slot_semaphore: asyncio.Semaphore,
        release_slot: Callable[[], None],
    ) -> Response:
        resolved_last_event_id = _resolve_last_event_id(request, last_event_id)
        channel = progress_channel(_schema, job_id)

        # ------------------------------------------------------------------
        # Phase 1: subscribe-before-query.
        #
        # Both the Redis subscribe and the PG query run in the handler body
        # (before EventSourceResponse is constructed) so that 404/503 errors
        # are returned as proper HTTP status codes rather than appearing mid-
        # stream after a 200 has already been sent.
        # ------------------------------------------------------------------

        pubsub = _redis_client.pubsub()
        try:
            await pubsub.subscribe(channel)
        except Exception as exc:
            logger.warning(
                "sse-redis-subscribe-failed",
                job_id=str(job_id),
                channel=channel,
                error=str(exc),
            )
            # Why bounded: same close contract as the generator finally —
            # helper never raises (suppress dropped), hung broker cannot
            # wedge the 503 path.
            await close_redis_bounded(pubsub, "web-progress", CLOSE_TIMEOUT_SECS)
            release_slot()
            return JSONResponse(
                status_code=503,
                content=_REDIS_503_BODY,
                headers={"Retry-After": "2"},
            )

        # short-lived PG connection — released before any SSE
        # byte is written.
        try:
            async with _pg_pool.acquire() as conn:
                row = await conn.fetchrow(_progress_sql, job_id)
        except Exception:
            # Cleanup must not mask the original exception from the PG query.
            with contextlib.suppress(Exception):
                await pubsub.unsubscribe(channel)
            # Why bounded: helper never raises (suppress dropped), so the
            # original PG error always propagates even with a dead broker.
            await close_redis_bounded(pubsub, "web-progress", CLOSE_TIMEOUT_SECS)
            raise

        if row is None:
            with contextlib.suppress(Exception):
                await pubsub.unsubscribe(channel)
            # Why bounded: helper never raises (suppress dropped), so the 404
            # is raised even with a dead broker.
            await close_redis_bounded(pubsub, "web-progress", CLOSE_TIMEOUT_SECS)
            raise HTTPException(status_code=404, detail="job not found")

        # Extract snapshot data from PG row.
        raw_progress_state: Any = row["progress_state"]
        progress_seq: int = row["progress_seq"]
        status: str = row["status"]
        is_terminal = status in TERMINAL_STATUSES
        progress_data = _serialize_progress_state(raw_progress_state)

        # ------------------------------------------------------------------
        # Phase 2: build and return EventSourceResponse.
        #
        # The generator owns pubsub from here; the try/finally inside
        # _event_generator ensures cleanup even on client disconnect
        # (CancelledError).
        # ------------------------------------------------------------------

        return EventSourceResponse(
            content=_event_generator(
                pubsub=pubsub,
                channel=channel,
                job_id=job_id,
                is_terminal=is_terminal,
                progress_seq=progress_seq,
                progress_data=progress_data,
                resolved_last_event_id=resolved_last_event_id,
                sse_slot_semaphore=sse_slot_semaphore,
                heartbeat_secs=_heartbeat_secs,
            ),
            headers=_SSE_HEADERS,
            # Effectively disable sse-starlette's built-in ping; we emit our own
            # keepalive comments via the get_message timeout loop.  ping=0
            # causes a tight loop (anyio.sleep(0) returns immediately), so we
            # use a 24-hour interval that will never fire in practice.
            ping=86_400,
            sep=_SSE_SEPARATOR,
        )

    # ----------------------------------------------------------------
    # Poll-state endpoint
    # ----------------------------------------------------------------

    @router.get("/api/job/{job_id}/state")
    async def job_state(  # pyright: ignore[reportUnusedFunction]  # Why: registered via FastAPI decorator.
        job_id: UUID,
    ) -> JSONResponse:
        """Return the current progress state for a job (polling fallback).

        Response body::

            {"status": <str>, "progress_state": <dict | null>, "progress_seq": <int>}

        HTTP 404 — job not found.
        """
        async with _pg_pool.acquire() as conn:
            row = await conn.fetchrow(_progress_sql, job_id)

        if row is None:
            raise HTTPException(status_code=404, detail="job not found")

        raw_ps: Any = row["progress_state"]
        progress_state: dict[str, object] | None
        if raw_ps is None:
            progress_state = None
        elif isinstance(raw_ps, dict):
            progress_state = cast("dict[str, object]", raw_ps)
        else:
            # asyncpg may return a str for jsonb; parse it.
            parsed: Any = _json.loads(raw_ps)
            progress_state = cast("dict[str, object]", parsed) if isinstance(parsed, dict) else None

        return JSONResponse(
            content={
                "status": row["status"],
                "progress_state": progress_state,
                "progress_seq": row["progress_seq"],
            }
        )

    return router

Admin UI router factory

_factory

Admin UI router factory: Jinja2 setup, auth hook, route registration.

Importing this module requires the taskq[fastapi] optional extra.

logger module-attribute

logger = structlog.get_logger('taskq.web.admin')

GZipStaticOnly

Bases: GZipMiddleware

GZip only static assets (/static/*), not HTML or JSON responses.

__call__ async

__call__(
    scope: Scope, receive: Receive, send: Send
) -> None

Source code in src/taskq/web/admin/_factory.py

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    if scope["type"] == "http":
        path: str = scope.get("path", "")
        if "/static/" not in path:
            await self.app(scope, receive, send)
            return
    await super().__call__(scope, receive, send)

AdminBundle dataclass

AdminBundle(
    router: APIRouter,
    templates: Environment,
    pg_pool: Pool,
    schema: str,
    redis_client: Any | None,
    settings: TaskQSettings,
    base_path: str,
    backend: Backend | None = None,
    rate_limit_registry: RateLimitRegistry | None = None,
)

Returned by create_router(); contains the router and all app.state values.

Pass this to setup_admin_state(app, bundle) in your lifespan before the first request, then mount bundle.router via app.include_router.

rate_limit_registry scopes the registry the rate-limit/reservation pages read; None resolves to the module singleton (the default — same-process behavior is unchanged).

router instance-attribute

router: APIRouter

templates instance-attribute

templates: Environment

pg_pool instance-attribute

pg_pool: Pool

schema instance-attribute

schema: str

redis_client instance-attribute

redis_client: Any | None

settings instance-attribute

settings: TaskQSettings

base_path instance-attribute

base_path: str

backend class-attribute instance-attribute

backend: Backend | None = None

rate_limit_registry class-attribute instance-attribute

rate_limit_registry: RateLimitRegistry | None = None

refresh_db_clock_offset async

refresh_db_clock_offset(pool: Pool) -> None

Re-measure the app-to-database clock offset, at most once per TTL.

Installed as a router-level dependency so every admin request keeps the offset fresh for the (synchronous) Jinja filter. Failures are swallowed and the previous offset kept: a clock probe must never take down a page, and a slightly stale offset is still far closer to the truth than ignoring skew entirely.

Source code in src/taskq/web/admin/_factory.py

async def refresh_db_clock_offset(pool: asyncpg.Pool) -> None:
    """Re-measure the app-to-database clock offset, at most once per TTL.

    Installed as a router-level dependency so every admin request keeps the
    offset fresh for the (synchronous) Jinja filter. Failures are swallowed
    and the previous offset kept: a clock probe must never take down a page,
    and a slightly stale offset is still far closer to the truth than
    ignoring skew entirely.
    """
    now = time.monotonic()
    if now < _db_clock_offset.expires_at:
        return
    try:
        before = datetime.now(UTC)
        async with pool.acquire() as conn:
            db_now: datetime = await conn.fetchval("SELECT clock_timestamp()")
        after = datetime.now(UTC)
    except Exception:
        # Back off for a full TTL rather than probing on every request.
        _db_clock_offset.expires_at = now + _CLOCK_OFFSET_TTL
        return
    # Why the midpoint: the round trip happens between the two local reads, so
    # the server's instant is best compared against the middle of that window
    # rather than either end — the same correction NTP applies. On a local
    # database this is sub-millisecond, but it costs nothing and keeps the
    # measurement honest over a slow link.
    app_now = before + (after - before) / 2
    _db_clock_offset.seconds = (db_now - app_now).total_seconds()
    _db_clock_offset.expires_at = now + _CLOCK_OFFSET_TTL

get_realtime_mode async

get_realtime_mode(
    redis_client: Any | None,
) -> tuple[str, str]

Return (realtime_mode, mode_label) using a 5 s server-side cache.

realtime_mode ∈ {"realtime", "polling", "polling-degraded"}. Cache is module-level — one entry covers all admin UI routes on the same process.

Source code in src/taskq/web/admin/_factory.py

async def get_realtime_mode(
    redis_client: Any | None,
) -> tuple[str, str]:
    """Return ``(realtime_mode, mode_label)`` using a 5 s server-side cache.

    realtime_mode ∈ {"realtime", "polling", "polling-degraded"}.
    Cache is module-level — one entry covers all admin UI routes
    on the same process.
    """
    if redis_client is None:
        return "polling", "polling mode"
    now = asyncio.get_running_loop().time()
    if now < _redis_health_cache.expires_at:
        ok = _redis_health_cache.ok
    else:
        try:
            await asyncio.wait_for(redis_client.ping(), timeout=0.5)
            ok = True
        except Exception:
            ok = False
        _redis_health_cache.ok = ok
        _redis_health_cache.expires_at = now + _CACHE_TTL
    if ok:
        return "realtime", "real-time mode"
    return "polling-degraded", "polling mode (Redis unavailable)"

get_pg_pool

get_pg_pool(request: Request) -> asyncpg.Pool

Dependency: yields the asyncpg pool from app.state.

Source code in src/taskq/web/admin/_factory.py

def get_pg_pool(request: Request) -> asyncpg.Pool:
    """Dependency: yields the asyncpg pool from ``app.state``."""
    pool: asyncpg.Pool = request.app.state.pg_pool
    return pool

get_backend

get_backend(request: Request) -> Backend | None

Dependency: yields the Backend from app.state if configured.

Source code in src/taskq/web/admin/_factory.py

def get_backend(request: Request) -> Backend | None:
    """Dependency: yields the Backend from ``app.state`` if configured."""
    return getattr(request.app.state, "backend", None)

get_rl_registry

get_rl_registry(request: Request) -> RateLimitRegistry

Dependency: yields the RateLimitRegistry from app.state.

setup_admin_state always sets the key (bundle instance or singleton). The getattr fallback keeps hand-assembled app.state setups (which never set the key) working exactly as today: the module singleton. Internal — deliberately not exported.

Source code in src/taskq/web/admin/_factory.py

def get_rl_registry(request: Request) -> RateLimitRegistry:
    """Dependency: yields the RateLimitRegistry from ``app.state``.

    ``setup_admin_state`` always sets the key (bundle instance or
    singleton). The ``getattr`` fallback keeps hand-assembled
    ``app.state`` setups (which never set the key) working exactly as
    today: the module singleton. Internal — deliberately not exported.
    """
    rl: RateLimitRegistry | None = getattr(request.app.state, "rate_limit_registry", None)
    return rl if rl is not None else _rl_singleton

get_schema

get_schema(request: Request) -> str

Dependency: yields the schema name from app.state.

Re-validates against :data:_IDENT_RE as defence-in-depth — the schema was validated at create_router construction time, but this ensures a runtime mutation of app.state.schema (e.g. by a misconfigured test fixture) cannot reach SQL interpolation.

Source code in src/taskq/web/admin/_factory.py

def get_schema(request: Request) -> str:
    """Dependency: yields the schema name from ``app.state``.

    Re-validates against :data:`_IDENT_RE` as defence-in-depth — the schema
    was validated at ``create_router`` construction time, but this ensures a
    runtime mutation of ``app.state.schema`` (e.g. by a misconfigured test
    fixture) cannot reach SQL interpolation.
    """
    s: str = request.app.state.schema
    if not _IDENT_RE.match(s):
        raise HTTPException(status_code=500, detail="invalid schema configuration")
    return s

get_redis_client

get_redis_client(request: Request) -> Any | None

Dependency: yields the redis client from app.state.

Source code in src/taskq/web/admin/_factory.py

def get_redis_client(request: Request) -> Any | None:
    """Dependency: yields the redis client from ``app.state``."""
    client: Any | None = request.app.state.redis_client
    return client

get_templates

get_templates(request: Request) -> Environment

Dependency: yields the Jinja2 Environment from app.state.

Source code in src/taskq/web/admin/_factory.py

def get_templates(request: Request) -> Environment:
    """Dependency: yields the Jinja2 Environment from ``app.state``."""
    env: Environment = request.app.state.templates
    return env

get_settings

get_settings(request: Request) -> TaskQSettings

Dependency: yields the TaskQSettings from app.state.

Source code in src/taskq/web/admin/_factory.py

def get_settings(request: Request) -> TaskQSettings:
    """Dependency: yields the TaskQSettings from ``app.state``."""
    s: TaskQSettings = request.app.state.settings
    return s

get_realtime_ctx async

get_realtime_ctx(
    redis_client: Any = Depends(get_redis_client),
) -> tuple[str, str]

Dependency: returns (realtime_mode, mode_label) for template rendering.

Source code in src/taskq/web/admin/_factory.py

async def get_realtime_ctx(
    redis_client: Any = Depends(get_redis_client),
) -> tuple[str, str]:
    """Dependency: returns (realtime_mode, mode_label) for template rendering."""
    return await get_realtime_mode(redis_client)

get_base_path

get_base_path(request: Request) -> str

Dependency: yields the admin UI base path from app.state.

Source code in src/taskq/web/admin/_factory.py

def get_base_path(request: Request) -> str:
    """Dependency: yields the admin UI base path from ``app.state``."""
    s: str = request.app.state.base_path
    return s

get_csrf_token

get_csrf_token(request: Request) -> str

Dependency: returns the CSRF token.

Prefers the token set by _CsrfRoute via request.state so the form hidden field and the cookie always carry the same value. Falls back to the cookie (present from a prior GET), then generates a fresh token.

Source code in src/taskq/web/admin/_factory.py

def get_csrf_token(request: Request) -> str:
    """Dependency: returns the CSRF token.

    Prefers the token set by ``_CsrfRoute`` via ``request.state``
    so the form hidden field and the cookie always carry the same value.
    Falls back to the cookie (present from a prior GET), then generates
    a fresh token.
    """
    token = getattr(request.state, "_csrf_token", None)
    if token is not None:
        return token
    return request.cookies.get(_CSRF_COOKIE_NAME) or secrets.token_hex(32)

validate_csrf async

validate_csrf(request: Request) -> None

Dependency: validates the synchronizer-token CSRF on POST requests.

Source code in src/taskq/web/admin/_factory.py

async def validate_csrf(request: Request) -> None:
    """Dependency: validates the synchronizer-token CSRF on POST requests."""
    cookie_token = request.cookies.get(_CSRF_COOKIE_NAME)
    if cookie_token is None:
        raise HTTPException(status_code=403, detail="CSRF token missing from cookies")
    form = await request.form()
    form_token = form.get("csrf_token")
    if not isinstance(form_token, str):
        raise HTTPException(status_code=403, detail="CSRF token missing from form")
    if not hmac.compare_digest(cookie_token, form_token):
        raise HTTPException(status_code=403, detail="CSRF token mismatch")

setup_admin_state

setup_admin_state(
    app: _AppLike, bundle: AdminBundle
) -> None

Populate app.state from bundle so route handler dependencies resolve.

Call this in your FastAPI lifespan after creating the bundle and before the first request arrives.

Source code in src/taskq/web/admin/_factory.py

def setup_admin_state(app: _AppLike, bundle: AdminBundle) -> None:
    """Populate ``app.state`` from *bundle* so route handler dependencies resolve.

    Call this in your FastAPI lifespan after creating the bundle and before
    the first request arrives.
    """
    app.state.pg_pool = bundle.pg_pool
    app.state.schema = bundle.schema
    app.state.redis_client = bundle.redis_client
    app.state.templates = bundle.templates
    app.state.settings = bundle.settings
    app.state.base_path = bundle.base_path
    app.state.backend = bundle.backend
    app.state.rate_limit_registry = (
        bundle.rate_limit_registry if bundle.rate_limit_registry is not None else _rl_singleton
    )

create_router

create_router(
    pg_pool: Pool,
    *,
    schema: str = "taskq",
    redis_client: Any | None = None,
    auth_dependency: Callable[..., Any] | None = None,
    base_path: str = "",
    backend: Backend | None = None,
    rate_limit_registry: RateLimitRegistry | None = None,
) -> AdminBundle

Create the admin UI FastAPI router.

Route handlers access shared resources (pool, schema, redis, settings, templates) via Depends(get_pg_pool) etc., which read from request.app.state. Call setup_admin_state(app, bundle) in your lifespan to populate those keys, then mount bundle.router at your chosen prefix via app.include_router.

base_path must match the prefix passed to include_router (e.g. "/admin"). It is injected as a Jinja2 global so templates can build prefix-safe URLs with {{ base_path }}/queues etc.

rate_limit_registry is an optional owned :class:RateLimitRegistry the admin pages read configured primitives from (e.g. the API-process instance in a multi-process deployment). Default None resolves to the module singleton — same-process behavior is unchanged.

Source code in src/taskq/web/admin/_factory.py

def create_router(
    pg_pool: asyncpg.Pool,
    *,
    schema: str = "taskq",
    redis_client: Any
    | None = None,  # Why: redis is an optional dependency (taskq[redis]); only runtime use is `is not None` boolean check — erasure boundary documented per erasure-boundary policy
    auth_dependency: Callable[..., Any] | None = None,
    base_path: str = "",
    backend: Backend | None = None,
    rate_limit_registry: RateLimitRegistry | None = None,
) -> AdminBundle:
    """Create the admin UI FastAPI router.

    Route handlers access shared resources (pool, schema, redis, settings,
    templates) via ``Depends(get_pg_pool)`` etc., which read from
    ``request.app.state``.  Call ``setup_admin_state(app, bundle)`` in your
    lifespan to populate those keys, then mount ``bundle.router`` at your
    chosen prefix via ``app.include_router``.

    ``base_path`` must match the prefix passed to ``include_router`` (e.g.
    ``"/admin"``).  It is injected as a Jinja2 global so templates can build
    prefix-safe URLs with ``{{ base_path }}/queues`` etc.

    ``rate_limit_registry`` is an optional owned :class:`RateLimitRegistry`
    the admin pages read configured primitives from (e.g. the API-process
    instance in a multi-process deployment).  Default ``None`` resolves to
    the module singleton — same-process behavior is unchanged.
    """
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema identifier: {schema!r}")

    settings = TaskQSettings.load()

    env = Environment(
        autoescape=True,
        loader=PackageLoader("taskq.web", "templates"),
    )
    env.globals["base_path"] = base_path  # pyright: ignore[reportArgumentType]  # Why: Jinja2 Environment.globals accepts arbitrary values for template globals; str is valid.
    env.globals["poll_interval_ms"] = int(settings.admin_ui_polling_interval_seconds * 1000)  # pyright: ignore[reportArgumentType]  # Why: same as above; int is a valid template global.
    env.filters["time_ago"] = _time_ago
    env.filters["iso_attr"] = _iso_attr

    # Why router-level: the relative-time filter is synchronous and cannot
    # query, so the app-to-database clock offset it needs has to be refreshed
    # by something that can. Every admin route serves timestamps, so the
    # dependency belongs on the router rather than being repeated per page --
    # and repeating it per page is how one page would end up telling a
    # different story about staleness than the next. It is cached for
    # _CLOCK_OFFSET_TTL, so this is one extra query per 30 s, not per request.
    router_dependencies: list[Any] = [Depends(_refresh_clock_offset)]
    if auth_dependency is not None:
        router_dependencies.insert(0, Depends(auth_dependency))
    router_kwargs: dict[str, Any] = {
        "route_class": _csrf_route_class(settings),
        "dependencies": router_dependencies,
    }

    router = APIRouter(**router_kwargs)

    if auth_dependency is None:
        is_dev_env = settings.environment in {"dev", "development"}
        if not is_dev_env and settings.admin_ui_require_auth:
            raise RuntimeError(
                "admin UI requires auth_dependency in non-dev environments "
                "(set TASKQ_ADMIN_UI_REQUIRE_AUTH=false to disable)"
            )
        # Why this warning sits outside the environment test that governs the
        # RuntimeError above: a dev-labeled process is the only configuration
        # that actually serves an unauthenticated admin UI, so it is the one
        # that most needs a log line. Keeping the warning inside the non-dev
        # branch meant the silent case was the dangerous one.
        suppressed_by = (
            "TASKQ_ENVIRONMENT is a dev environment, so the fail-closed startup check did not run"
            if is_dev_env
            else "TASKQ_ADMIN_UI_REQUIRE_AUTH is false, so the fail-closed "
            "startup check was suppressed"
        )
        logger.warning(
            "admin-ui-no-auth",
            environment=settings.environment,
            detail=(
                "admin UI is being served with no authentication: "
                f"{suppressed_by}. Every admin route is reachable by anyone "
                "who can reach this port. This is unsafe if the process is "
                "actually serving production traffic: a production "
                "deployment mislabeled as dev disables this check and the "
                "health/metrics token check (TASKQ_HEALTH_REQUIRE_TOKEN) at "
                "the same time. Pass auth_dependency to create_router, or "
                "set TASKQ_ENVIRONMENT to the real environment so startup "
                "fails closed."
            ),
        )

    @router.get("/")
    async def index() -> RedirectResponse:  # pyright: ignore[reportUnusedFunction]  # Why: registered via FastAPI decorator; pyright cannot see the route registration.
        return RedirectResponse(url="queues", status_code=302)

    _static.register(router, _STATIC_DIR)

    _discover_and_register(router)

    # ── Progress SSE / poll-state routes ────────────────────────────────
    # The admin UI's realtime.js connects to these endpoints for live
    # progress streaming.  Mount at /jobs so the paths become
    #   /jobs/api/job/{job_id}/progress/stream   (SSE)
    #   /jobs/api/job/{job_id}/state             (poll-state JSON)
    from taskq.web.progress import create_router as _create_progress_router

    progress_router = _create_progress_router(
        pg_pool,
        redis_client,
        schema=schema,
        auth_dependency=auth_dependency,
    )
    router.include_router(progress_router, prefix="/jobs")

    return AdminBundle(
        router=router,
        templates=env,
        pg_pool=pg_pool,
        schema=schema,
        redis_client=redis_client,
        settings=settings,
        base_path=base_path,
        backend=backend,
        rate_limit_registry=rate_limit_registry,
    )

Health router

health

FastAPI router for /jobs/health/{live,ready}.

GET /jobs/health/metrics is served by taskq.contrib.prometheus.create_metrics_router (requires taskq[prometheus]) and must be mounted alongside this router.

Importing this module requires the taskq[fastapi] optional extra.

logger module-attribute

logger = structlog.get_logger('taskq.web.health')

create_health_router

create_health_router(deps: WorkerDeps) -> APIRouter

Create a FastAPI router at /jobs/health/{live,ready}.

Captures deps via closure — no FastAPI dependency injection. Mount alongside create_metrics_router (taskq.contrib.prometheus) for the full /jobs/health surface including Prometheus metrics.

Source code in src/taskq/web/health.py

def create_health_router(deps: "WorkerDeps") -> APIRouter:
    """Create a FastAPI router at /jobs/health/{live,ready}.

    Captures *deps* via closure — no FastAPI dependency injection.
    Mount alongside create_metrics_router (taskq.contrib.prometheus) for the
    full /jobs/health surface including Prometheus metrics.
    """

    router = APIRouter(prefix="/jobs/health")

    @router.get("/live")
    async def live() -> Response:  # pyright: ignore[reportUnusedFunction]  # Why: registered via FastAPI decorator; pyright cannot see the route registration.
        t0 = time.perf_counter()
        ok, _msg = await _check_live()
        status_code = 200 if ok else 503
        body_dict: dict[str, str] = {"status": "ok"} if ok else {"status": "unresponsive"}
        body_bytes = _json.dumps(body_dict)

        elapsed_ms = (time.perf_counter() - t0) * 1000.0
        logger.debug(
            "health-request",
            endpoint="/jobs/health/live",
            status_code=status_code,
            response_time_ms=elapsed_ms,
        )

        return Response(
            content=body_bytes,
            media_type="application/json",
            status_code=status_code,
        )

    @router.get("/ready")
    async def ready() -> Response:  # pyright: ignore[reportUnusedFunction]  # Why: registered via FastAPI decorator; pyright cannot see the route registration.
        t0 = time.perf_counter()
        report = await compute_health(deps)
        body_bytes = build_ready_body(report, deps)
        status_code = 200 if report.ready else 503

        elapsed_ms = (time.perf_counter() - t0) * 1000.0
        logger.debug(
            "health-request",
            endpoint="/jobs/health/ready",
            status_code=status_code,
            response_time_ms=elapsed_ms,
        )

        return Response(
            content=body_bytes,
            media_type="application/json",
            status_code=status_code,
        )

    return router