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).
• Each broker read carries an app-level deadline: the delegated timeout= bounds only the read itself, while a reconnect inside redis-py's parse_response (broker dropped mid-read) is bounded by nothing, so the loop wraps every read in asyncio.wait_for and a read that outlives the deadline ends the stream, exactly as broker death does.
• On client disconnect, try/finally in the generator calls pubsub.unsubscribe() and pubsub.aclose() to prevent stale Redis subscriptions.
• Session re-check (#316): the router-level Depends(auth_dependency) runs once at request acceptance; a stream that authenticated once would otherwise keep delivering frames long after its session was revoked. The generator therefore re-invokes an optional session_verifier -- the re-check the auth dependency itself exposes (create_auth_dependency attaches one) -- before every yielded event and at every keepalive tick, and ends the stream on failure. The finally releases the SSE slot and the Redis subscription. Bounded staleness: at most one keepalive interval between the revocation and the stream ending; the effective keepalive interval is capped at _SSE_HEARTBEAT_CAP_SECS so the bound holds no matter how a host configures sse_heartbeat_interval (the clamp is logged at startup). The re-check itself is bounded too (SESSION_RECHECK_TIMEOUT_SECS): a verifier that hangs is fail-closed revocation, never a frozen generator. The check is cookie-only -- no session store, no DB round trip -- so server-side revocation of a still-validly-signed cookie is impossible by construction (docs/guides/sso.md states the same limits).

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,
    resolve_pg_pool: Callable[[Request], Pool]
    | None = None,
    resolve_redis_client: Callable[[Request], Any]
    | None = None,
    session_verifier: Callable[[Request], Awaitable[bool]]
    | 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). Outside a dev environment (TASKQ_ENVIRONMENT not dev/development) the factory raises RuntimeError when it is omitted, unless TASKQ_PROGRESS_REQUIRE_AUTH=false suppresses the check; serving without auth always logs a warning. 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. resolve_pg_pool / resolve_redis_client: Per-request resolvers for the pool and the Redis client, used instead of pg_pool / redis_client when given. A host whose pool is replaced while it runs - taskq ui serve rebuilds it on a credential rotation - resolves the live one from its own state on every request rather than serving from the pool this router was constructed with, which that rotation has closed. Each is a FastAPI dependency: it may declare request: Request. session_verifier: Optional async re-check for long-lived SSE streams (#316): Callable[[Request], Awaitable[bool]] returning whether the session that opened the stream is still valid. Live streams re-invoke it before every yielded event and at every keepalive tick and end on failure, so a session revoked mid-stream (secret rotation, expiry, an allowlist change) stops receiving frames within one keepalive interval, capped at 60 s. Each invocation is bounded by SESSION_RECHECK_TIMEOUT_SECS (5 s): a verifier that hangs or outlives the bound is treated as failure -- fail closed -- so a wedged host verifier cannot freeze the generator inside its keepalive path. When omitted, the router derives it from the session_verifier attribute the taskq auth dependencies (create_auth_dependency, token_auth) attach to the callable they return. A host supplying its own auth_dependency MUST pass session_verifier explicitly to keep the re-check: without the attribute the router logs a one-per-router progress-stream-no-session-verifier warning and the streams authenticate once, with no per-frame re-check.

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,
    resolve_pg_pool: Callable[[Request], asyncpg.Pool] | None = None,
    resolve_redis_client: Callable[[Request], Any] | None = None,
    session_verifier: Callable[[Request], Awaitable[bool]] | 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``). Outside a dev environment
        (``TASKQ_ENVIRONMENT`` not ``dev``/``development``) the factory
        raises ``RuntimeError`` when it is omitted, unless
        ``TASKQ_PROGRESS_REQUIRE_AUTH=false`` suppresses the check; serving
        without auth always logs a warning.
    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.
    resolve_pg_pool / resolve_redis_client:
        Per-request resolvers for the pool and the Redis client, used
        instead of *pg_pool* / *redis_client* when given. A host whose pool
        is replaced while it runs - ``taskq ui serve`` rebuilds it on a
        credential rotation - resolves the live one from its own state on
        every request rather than serving from the pool this router was
        constructed with, which that rotation has closed. Each is a
        FastAPI dependency: it may declare ``request: Request``.
    session_verifier:
        Optional async re-check for long-lived SSE streams
        (#316): ``Callable[[Request], Awaitable[bool]]`` returning whether the
        session that opened the stream is still valid. Live streams re-invoke
        it before every yielded event and at every keepalive tick and end on
        failure, so a session revoked mid-stream (secret rotation, expiry, an
        allowlist change) stops receiving frames within one keepalive
        interval, capped at 60 s. Each invocation is bounded by
        ``SESSION_RECHECK_TIMEOUT_SECS`` (5 s): a verifier that hangs or
        outlives the bound is treated as failure -- fail closed -- so a wedged
        host verifier cannot freeze the generator inside its keepalive path.
        When omitted, the router derives it from the ``session_verifier``
        attribute the taskq auth dependencies (``create_auth_dependency``,
        ``token_auth``) attach to the callable they return. A host supplying
        its own ``auth_dependency`` MUST pass ``session_verifier`` explicitly
        to keep the re-check: without the attribute the router logs a
        one-per-router ``progress-stream-no-session-verifier`` warning and
        the streams authenticate once, with no per-frame re-check.
    """
    require_schema(schema)

    settings = TaskQSettings.load()

    if auth_dependency is None:
        if not settings.is_dev_environment and settings.progress_require_auth:
            raise RuntimeError(
                "progress router requires auth_dependency in non-dev environments "
                "(set TASKQ_PROGRESS_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 progress router, 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 settings.is_dev_environment
            else "TASKQ_PROGRESS_REQUIRE_AUTH is false, so the fail-closed "
            "startup check was suppressed"
        )
        logger.warning(
            "progress-router-no-auth",
            environment=settings.environment,
            detail=(
                "the progress router is being served with no authentication: "
                f"{suppressed_by}. The SSE stream and per-job state endpoints "
                "are reachable by anyone who can reach this port, and each "
                "stream holds a Redis pubsub subscription and an asyncio task "
                "for as long as the client stays connected. Pass "
                "auth_dependency to create_router, or set TASKQ_ENVIRONMENT "
                "to the real environment so startup fails closed."
            ),
        )

    router_kwargs: dict[str, Any] = {
        "tags": ["progress"],
        # F6: the monitor's HEAD check against the stream/state routes
        # must not 405 (the same route class the admin router's pages use).
        "route_class": HeadForGetRoute,
    }
    if auth_dependency is not None:
        router_kwargs["dependencies"] = [Depends(auth_dependency)]

    router = APIRouter(**router_kwargs)

    # Session re-check derivation (#316): prefer the explicit parameter, fall
    # back to the attribute the taskq auth dependencies attach to the callable
    # they return. An auth dependency without a re-check cannot be safely
    # re-invoked by us (it may need FastAPI dependency injection), so the
    # honest fallback is a loud warning, not a silent best-effort call.
    if session_verifier is None and auth_dependency is not None:
        derived: Any = getattr(auth_dependency, "session_verifier", None)
        session_verifier = cast("Callable[[Request], Awaitable[bool]] | None", derived)
    if auth_dependency is not None and session_verifier is None:
        logger.warning(
            "progress-stream-no-session-verifier",
            detail=(
                "auth_dependency exposes no session_verifier re-check, so SSE "
                "streams authenticate once at subscribe and will keep "
                "delivering frames after a session is revoked. Pass a "
                "session_verifier to create_router, or build the dependency "
                "with taskq's create_auth_dependency/token_auth, which attach "
                "one."
            ),
        )

    _schema = schema
    _heartbeat_secs = sse_heartbeat_interval.total_seconds()
    # Why the cap: the keepalive tick is the revocation bound (#316) -- the
    # generator re-checks the session once per tick -- so a host-configured
    # interval of hours would silently move that bound to hours. The loop and
    # the fallback ping below are both capped at _SSE_HEARTBEAT_CAP_SECS.
    _effective_heartbeat_secs = min(_heartbeat_secs, _SSE_HEARTBEAT_CAP_SECS)
    if _heartbeat_secs > _SSE_HEARTBEAT_CAP_SECS:
        # The clamp itself must not be silent: a host that asked for an hour
        # between keepalives otherwise discovers the 60 s floor only by
        # reading the source. Once per router (this runs in create_router).
        logger.warning(
            "sse-heartbeat-interval-clamped",
            configured_seconds=_heartbeat_secs,
            effective_seconds=_effective_heartbeat_secs,
            detail=(
                "sse_heartbeat_interval exceeds the "
                f"{_SSE_HEARTBEAT_CAP_SECS:g} s cap and is clamped: the "
                "keepalive tick is also the session re-check cadence "
                "(#316), so a longer interval would silently loosen the "
                "bound on how long a revoked session keeps receiving "
                "frames."
            ),
        )
    _acquire_timeout = settings.admin_acquire_timeout

    # The degraded-mode log fires once per router, not per request: the
    # first stream a misconfigured portal refuses names the condition,
    # every later 503 is the same fact (a per-request warning is a log
    # flood that gets filtered out, which is the same as being silent).
    _degraded_mode_logged = False

    def _constructed_pool() -> asyncpg.Pool:
        return pg_pool

    def _constructed_redis() -> Any:
        return redis_client

    _resolve_pool: Callable[..., Any] = (
        resolve_pg_pool if resolve_pg_pool is not None else _constructed_pool
    )

    def _get_pool(pool: asyncpg.Pool = Depends(_resolve_pool)) -> BoundedPool:
        # Every checkout below is bounded (TASKQ_ADMIN_ACQUIRE_TIMEOUT): a
        # pool with nothing to give answers 503 instead of hanging the
        # request and every request behind it.
        return BoundedPool(pool, acquire_timeout=_acquire_timeout, role="progress")

    _get_redis: Callable[..., Any] = (
        resolve_redis_client if resolve_redis_client is not None else _constructed_redis
    )
    if max_sse_connections is None:
        max_sse_connections = settings.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,
        pg_pool: BoundedPool = Depends(_get_pool),
        redis_client: Any = Depends(_get_redis),
    ) -> 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.
        """
        nonlocal _degraded_mode_logged
        if redis_client is None:
            # One-time degraded-mode log (the operator-side companion of
            # the factory's ``admin-ui-no-redis-client`` startup warning):
            # this 503 is what puts the browser into polling mode, so the
            # log names it. Once per router; see the flag above.
            if not _degraded_mode_logged:
                _degraded_mode_logged = True
                logger.warning(
                    "progress-stream-polling-degraded",
                    detail=(
                        "the admin portal answered a progress stream with "
                        "no Redis client (503 redis_not_configured): the "
                        "dashboard falls back to polling the per-job state "
                        "endpoint. Set TASKQ_REDIS_URL and pass the client "
                        "to create_router to restore live SSE progress"
                    ),
                )
            return _OrjsonJSONResponse(  # pyright: ignore[reportReturnType]  # Why: FastAPI accepts any Response subclass here; the JSON response 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()
                # The gauge's matching close for the early-exit paths
                # (503 subscribe failure, PG error, 404): the slot and
                # the level move together here exactly as they do in the
                # generator's own finally on the streaming paths.
                record_sse_connection_closed("progress", "progress-stream")

        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,
                pg_pool=pg_pool,
                redis_client=redis_client,
                session_verifier=session_verifier,
            )
        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],
        pg_pool: BoundedPool,
        redis_client: Any,
        session_verifier: Callable[[Request], Awaitable[bool]] | 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 _OrjsonJSONResponse(
                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.
        #
        # Why the guard: the slot and the pubsub subscription are both live
        # in this third region - between the 503/PG-error/404 arms above and
        # the generator's finally taking ownership below. A raise here (a
        # host-installed custom jsonb codec surfacing through
        # _serialize_progress_state; unreachable with stock codecs, where
        # asyncpg hands back a str/dict the serializer never fails on)
        # propagates to progress_stream's release_slot with the subscription
        # still open: a leaked broker subscription per failed request. Every
        # early exit between subscribe and the generator has to give it
        # back, so the same cleanup runs here - suppressed and bounded, the
        # original exception still propagates.
        try:
            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).
            # ------------------------------------------------------------------

            # Bind the request into the re-check once: the verifier reads the
            # same cookie (or bearer token) the request arrived with, and a
            # session revoked mid-stream -- secret rotation, expiry, an
            # allowlist change -- fails the re-check even though those bytes
            # are unchanged.
            _session_verifier: Callable[[], Awaitable[bool]] | None = (
                (lambda: session_verifier(request)) if session_verifier is not None else None
            )

            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=_effective_heartbeat_secs,
                    session_verifier=_session_verifier,
                ),
                headers=_SSE_HEADERS,
                # We emit our own keepalive comments via the get_message timeout
                # loop; sse-starlette's ping is only a fallback. Bounded at twice
                # the effective (capped) heartbeat: the old 24-hour value was a
                # no-lifetime-cap stream -- exactly the property the #316 fix
                # removes -- while ping=0 causes a tight loop (anyio.sleep(0)
                # returns immediately).
                ping=_effective_heartbeat_secs * 2,
                sep=_SSE_SEPARATOR,
            )
        except BaseException:
            # Cleanup must not mask the original exception from the snapshot
            # region.
            with contextlib.suppress(Exception):
                await pubsub.unsubscribe(channel)
            # Why bounded: helper never raises (suppress dropped), so the
            # original error always propagates even with a dead broker.
            await close_redis_bounded(pubsub, "web-progress", CLOSE_TIMEOUT_SECS)
            raise

    # ----------------------------------------------------------------
    # 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,
        request: Request,
        pg_pool: BoundedPool = Depends(_get_pool),
    ) -> Response:
        """Return the current progress state for a job (polling fallback).

        Polling is a first-class mode of the admin UI, so the poll is
        conditional: a client that already rendered the current state
        sends ``If-None-Match`` with the progress sequence it has (the
        seq doubles as the ETag), and an unchanged tick is answered
        ``304`` with no body at all - the poll cadence never re-downloads
        data the client already has.

        Response body::

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

        HTTP 304, ``If-None-Match`` matches the current progress_seq: the
        client's rendered state is current, nothing to download.
        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")

        # The progress sequence is a monotonically increasing write cursor
        # on the progress bytes this endpoint returns, so it IS the ETag
        # for the progress state: a seq the client already rendered means
        # every progress field (including the fingerprint the client
        # derives from it) is unchanged.
        #
        # The terminal rows are the exception: the terminal writes' seq
        # expression is GREATEST(progress_seq, $buffer_seq) and the
        # consumer passes the seq its progress buffer just flushed,
        # exactly the row's current cursor, so the status flips terminal
        # and the cursor stands still. A 304 there would hide the
        # terminal transition behind an unchanged cursor, and the poll
        # client (realtime.js stops only on a DOWNLOADED terminal status)
        # would render running forever. A terminal row therefore always
        # answers 200: the terminal body is the LAST body a poll client
        # needs, the one download that ends the poll loop, so the cost is
        # one bounded download per job and the seq-matching 304 keeps
        # serving the steady non-terminal state.
        etag = f'"{row["progress_seq"]}"'
        is_terminal = row["status"] in TERMINAL_STATUSES
        if not is_terminal and request.headers.get("if-none-match") == etag:
            return Response(status_code=304, headers={"ETag": etag})

        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 _OrjsonJSONResponse(
            content={
                "status": row["status"],
                "progress_state": progress_state,
                "progress_seq": row["progress_seq"],
            },
            headers={"ETag": etag},
        )

    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)

ActorFirePolicyLike

Bases: Protocol

Structural twin of worker.cron_loop.ActorFirePolicy.

The admin package must stay importable without the worker (test_no_worker_import), so the run-now handler cannot name the cron loop's class at module level; this Protocol pins the two attributes the fire path needs. The cron loop's frozen dataclass satisfies it structurally, and so does any host-supplied mapping value with the same shape.

singleton property

singleton: bool

max_pending property

max_pending: int | None

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,
    actor_fire_policies: Mapping[str, ActorFirePolicyLike]
    | None = None,
    session_verifier: Callable[[Request], Awaitable[bool]]
    | 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

actor_fire_policies class-attribute instance-attribute

actor_fire_policies: (
    Mapping[str, ActorFirePolicyLike] | None
) = None

session_verifier class-attribute instance-attribute

session_verifier: (
    Callable[[Request], Awaitable[bool]] | None
) = None

refresh_db_clock_offset async

refresh_db_clock_offset(pool: BoundedPool) -> 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. A failed probe keeps the previous offset and is logged, once per back-off window, as admin-clock-offset-probe-failed: 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 - but every relative age on every page renders against it, so a probe that keeps failing has to be visible rather than silently ageing what the operator reads.

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

async def refresh_db_clock_offset(pool: BoundedPool) -> 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. A failed probe keeps
    the previous offset and is logged, once per back-off window, as
    ``admin-clock-offset-probe-failed``: 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 - but every relative age on every
    page renders against it, so a probe that keeps failing has to be
    visible rather than silently ageing what the operator reads.
    """
    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 as exc:
        # Back off for a full TTL rather than probing on every request; the
        # back-off is also what bounds this warning to one per window.
        _db_clock_offset.expires_at = now + _CLOCK_OFFSET_TTL
        logger.warning(
            "admin-clock-offset-probe-failed",
            error_type=type(exc).__name__,
            error=str(exc),
            retry_in_seconds=_CLOCK_OFFSET_TTL,
            kept_offset_seconds=_db_clock_offset.seconds,
        )
        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.

The raw pool - for the streams and library calls that manage their own checkouts. Request handlers that run queries take :class:BoundedPool through :func:get_admin_pool instead, whose every checkout is bounded.

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``.

    The raw pool - for the streams and library calls that manage their own
    checkouts. Request handlers that run queries take :class:`BoundedPool`
    through :func:`get_admin_pool` instead, whose every checkout is
    bounded.
    """
    pool: asyncpg.Pool = request.app.state.pg_pool
    return pool

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_admin_pool

get_admin_pool(
    pool: Pool = Depends(get_pg_pool),
    settings: TaskQSettings = Depends(get_settings),
) -> BoundedPool

Dependency: the admin pool with bounded checkouts (see :class:BoundedPool).

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

def get_admin_pool(
    pool: asyncpg.Pool = Depends(get_pg_pool),
    settings: TaskQSettings = Depends(get_settings),
) -> BoundedPool:
    """Dependency: the admin pool with bounded checkouts (see :class:`BoundedPool`)."""
    return BoundedPool(pool, acquire_timeout=settings.admin_acquire_timeout, role="admin")

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_actor_fire_policies

get_actor_fire_policies(
    request: Request,
) -> Mapping[str, ActorFirePolicyLike] | None

Dependency: the host's actor fire policies, if it supplied any.

setup_admin_state always sets the key when the bundle carried a mapping; the getattr fallback keeps hand-assembled app.state setups (which never set the key) working exactly as before: no policies, no stamping, the documented residual.

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

def get_actor_fire_policies(request: Request) -> Mapping[str, ActorFirePolicyLike] | None:
    """Dependency: the host's actor fire policies, if it supplied any.

    ``setup_admin_state`` always sets the key when the bundle carried a
    mapping; the ``getattr`` fallback keeps hand-assembled ``app.state``
    setups (which never set the key) working exactly as before: no
    policies, no stamping, the documented residual.
    """
    return getattr(request.app.state, "actor_fire_policies", 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_session_verifier

get_session_verifier(
    request: Request,
) -> Callable[[Request], Awaitable[bool]] | None

Dependency: the SSE session re-check from app.state (#316).

Long-lived SSE streams re-invoke it (before every yielded event, at every keepalive tick) so a session revoked mid-stream stops receiving frames. getattr with a None default rather than an attribute read: a host that mounted the router without calling setup_admin_state -- the state the other page dependencies already require -- gets an unrechecked stream (authenticate-once) rather than a 500 on every /sse route.

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

def get_session_verifier(request: Request) -> Callable[[Request], Awaitable[bool]] | None:
    """Dependency: the SSE session re-check from ``app.state`` (#316).

    Long-lived SSE streams re-invoke it (before every yielded event, at every
    keepalive tick) so a session revoked mid-stream stops receiving frames.
    ``getattr`` with a ``None`` default rather than an attribute read: a host
    that mounted the router without calling ``setup_admin_state`` -- the state
    the other page dependencies already require -- gets an unrechecked stream
    (authenticate-once) rather than a 500 on every ``/sse`` route.
    """
    verifier: Any = getattr(request.app.state, "taskq_session_verifier", None)
    return verifier

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_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")

require_actions_enabled async

require_actions_enabled(
    settings: TaskQSettings = Depends(get_settings),
) -> None

Dependency: refuses mutations on a deployment that disabled them.

Declared BEFORE validate_csrf on every mutation route (FastAPI resolves dependencies in declaration order): whether actions are enabled is a configuration state, safe to answer before any token validation -- an operator who POSTs to a disabled deployment gets the actionable config message, never a CSRF error that reads as a bug. The token still guards every route when actions ARE enabled (the enabled-check is a precondition, not a replacement).

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

async def require_actions_enabled(
    settings: TaskQSettings = Depends(get_settings),
) -> None:
    """Dependency: refuses mutations on a deployment that disabled them.

    Declared BEFORE ``validate_csrf`` on every mutation route (FastAPI
    resolves dependencies in declaration order): whether actions are
    enabled is a configuration state, safe to answer before any token
    validation -- an operator who POSTs to a disabled deployment gets the
    actionable config message, never a CSRF error that reads as a bug.
    The token still guards every route when actions ARE enabled (the
    enabled-check is a precondition, not a replacement).
    """
    if not settings.admin_actions_enabled:
        raise HTTPException(status_code=403, detail=_ACTIONS_DISABLED_DETAIL)

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.

This also installs the admin error handlers on the host app (U1/U2): the Postgres-connection failure family becomes a 503 (the JSON envelope {"detail": "postgres_unavailable"} with Retry-After for the machine routes, a branded self-retrying HTML page for the operator routes) and the HTML-wanting 4xxs (a missing job's 404, a refused filter's 400, an unparseable parameter's 422) render the error page instead of a JSON wall. The API-ish routes (/api/*, /jobs/count, /sse/*) always keep JSON, and a host route outside the admin base path is never touched.

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.

    This also installs the admin error handlers on the host app (U1/U2):
    the Postgres-connection failure family becomes a 503 (the JSON
    envelope ``{"detail": "postgres_unavailable"}`` with ``Retry-After``
    for the machine routes, a branded self-retrying HTML page for the
    operator routes) and the HTML-wanting 4xxs (a missing job's 404, a
    refused filter's 400, an unparseable parameter's 422) render the
    error page instead of a JSON wall. The API-ish routes (``/api/*``,
    ``/jobs/count``, ``/sse/*``) always keep JSON, and a host route
    outside the admin base path is never touched.
    """
    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
    )
    app.state.actor_fire_policies = bundle.actor_fire_policies
    app.state.taskq_session_verifier = bundle.session_verifier
    _install_admin_error_handlers(app, bundle)

get_principal

get_principal(request: Request) -> Any

Dependency: the principal the auth dependency returned for this request.

An :class:~taskq.web.admin.auth.IdentityClaims for the shipped SSO dependencies, whatever a host's custom auth_dependency returns, or None on the no-auth dev path (resolved to anonymous at the audit boundary). Mutation handlers pass it straight through to :func:taskq.web.admin._audit.record_admin_action.

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

def get_principal(request: Request) -> Any:
    """Dependency: the principal the auth dependency returned for this request.

    An :class:`~taskq.web.admin.auth.IdentityClaims` for the shipped SSO
    dependencies, whatever a host's custom ``auth_dependency`` returns, or
    ``None`` on the no-auth dev path (resolved to ``anonymous`` at the
    audit boundary). Mutation handlers pass it straight through to
    :func:`taskq.web.admin._audit.record_admin_action`.
    """
    return getattr(request.state, "principal", None)

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,
    actor_fire_policies: Mapping[str, ActorFirePolicyLike]
    | None = None,
    session_verifier: Callable[[Request], Awaitable[bool]]
    | 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.

actor_fire_policies is an optional mapping from actor name to the fire policy that actor's ActorRef declares (singleton, max_pending), the same mapping the worker bootstrap hands the cron tick. The run-now fire path consults it because the stored actor_config row cannot carry code-declared flags: without it a run-now fire for a singleton actor lands WITHOUT metadata["singleton"] -- invisible to the jobs_singleton_uniq index and to every later singleton preflight, the "at most one active job" contract silently gone from the admin surface. Pass the worker's mapping when the admin app is mounted in a process that has the registry (the worker itself); None keeps the residual: a standalone admin process cannot know, and the run-now fire stamps nothing (disclosed here and in ops.py, not silent).

session_verifier is the optional async re-check long-lived SSE streams re-invoke (#316) -- Callable[[Request], Awaitable[bool]] returning whether the session that opened the stream is still valid. When omitted, it is derived from the session_verifier attribute the taskq auth dependencies (create_auth_dependency, token_auth) attach to the callable they return. A host supplying its own auth_dependency without that attribute MUST pass session_verifier explicitly to keep the re-check; otherwise the admin /sse/{topic} streams authenticate once at subscribe with no mid-stream re-check, and a one-per-router admin-sse-no-session-verifier warning fires at startup.

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,
    actor_fire_policies: Mapping[str, ActorFirePolicyLike] | None = None,
    session_verifier: Callable[[Request], Awaitable[bool]] | 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.

    ``actor_fire_policies`` is an optional mapping from actor name to the
    fire policy that actor's ``ActorRef`` declares (singleton, max_pending),
    the same mapping the worker bootstrap hands the cron tick. The run-now
    fire path consults it because the stored ``actor_config`` row cannot
    carry code-declared flags: without it a run-now fire for a singleton
    actor lands WITHOUT ``metadata["singleton"]`` -- invisible to the
    ``jobs_singleton_uniq`` index and to every later singleton preflight,
    the "at most one active job" contract silently gone from the admin
    surface. Pass the worker's mapping when the admin app is mounted in a
    process that has the registry (the worker itself); ``None`` keeps the
    residual: a standalone admin process cannot know, and the run-now fire
    stamps nothing (disclosed here and in ops.py, not silent).

    ``session_verifier`` is the optional async re-check long-lived SSE streams
    re-invoke (#316) -- ``Callable[[Request], Awaitable[bool]]`` returning
    whether the session that opened the stream is still valid. When omitted,
    it is derived from the ``session_verifier`` attribute the taskq auth
    dependencies (``create_auth_dependency``, ``token_auth``) attach to the
    callable they return. A host supplying its own ``auth_dependency``
    without that attribute MUST pass ``session_verifier`` explicitly to keep
    the re-check; otherwise the admin ``/sse/{topic}`` streams authenticate
    once at subscribe with no mid-stream re-check, and a one-per-router
    ``admin-sse-no-session-verifier`` warning fires at startup.
    """
    require_schema(schema)

    settings = TaskQSettings.load()

    # Why this warning exists (the #424 pattern, admin side) and why it is
    # conditioned on the settings: TaskQ Redis configured but this router
    # created without the client is a WIRING GAP - the pages render
    # polling mode and the progress stream answers 503
    # ``redis_not_configured``, so live SSE progress is simply absent
    # while the dashboard keeps working, a degradation the operator
    # otherwise discovers only by staring at the mode badge. A deployment
    # with NO TaskQ Redis configured at all is the legitimate polling
    # mode: the badge says so accurately and no warning fires. Only the
    # gap (settings expect Redis, app state lacks the client) is loud.
    # Once per admin app: create_router runs once per router.
    if settings.redis_url is not None and redis_client is None:
        logger.warning(
            "admin-ui-no-redis-client",
            detail=(
                "the admin portal has no Redis client: live SSE progress "
                "is unavailable, the dashboard falls back to polling"
            ),
            remedy=(
                "TaskQ has Redis configured (TASKQ_REDIS_URL is set) but "
                "this admin router was created without the client: pass "
                "it as create_router(redis_client=...) so it reaches "
                "app.state via setup_admin_state (taskq ui serve wires it "
                "automatically); without it the per-job progress stream "
                "answers 503 redis_not_configured and every job page "
                "polls Postgres instead of streaming"
            ),
        )

    # Session re-check derivation (#316): prefer the explicit parameter, fall
    # back to the attribute the taskq auth dependencies attach. An auth
    # dependency without a re-check cannot be safely re-invoked by us (it may
    # need FastAPI dependency injection), so the fallback is no re-check --
    # the same subscribe-once model the router-level Depends alone gives.
    if session_verifier is None and auth_dependency is not None:
        derived_verifier: Any = getattr(auth_dependency, "session_verifier", None)
        session_verifier = cast("Callable[[Request], Awaitable[bool]] | None", derived_verifier)
    if auth_dependency is not None and session_verifier is None:
        # Same loud warning the progress router emits: a third-party auth
        # dependency (one built without taskq's create_auth_dependency /
        # token_auth) carries no session_verifier attribute, and the honest
        # fallback is streams that authenticate once -- which must be a
        # startup warning, not a silent best-effort call. Once per router:
        # this runs in create_router, before any stream exists.
        logger.warning(
            "admin-sse-no-session-verifier",
            detail=(
                "auth_dependency exposes no session_verifier re-check, so the "
                "admin /sse/{topic} streams authenticate once at subscribe "
                "and keep delivering frames after a session is revoked. Pass "
                "a session_verifier to create_router, or build the "
                "dependency with taskq's create_auth_dependency/token_auth, "
                "which attach one."
            ),
        )

    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.
    # Why a callable rather than a plain value: the token is per request (the
    # auth dependency arms it from the live SSO session cookie), so it cannot
    # be baked into the environment at startup. The chrome calls it to decide
    # whether the Sign out control renders at all.
    env.globals["sso_logout_token"] = current_sso_logout_token  # pyright: ignore[reportArgumentType]  # Why: Jinja2 Environment.globals accepts arbitrary values for template globals; a zero-arg callable is valid.
    # One source for the disabled-actions banner text: the same string the
    # mutation routes' 403 detail carries (require_actions_enabled), so a
    # page and a refusal never tell two different stories about the knob.
    env.globals["admin_actions_banner"] = _ACTIONS_DISABLED_DETAIL  # pyright: ignore[reportArgumentType]  # Why: Jinja2 Environment.globals accepts arbitrary values for template globals; str is valid.
    # The status colors' single source (see _constants.STATUS_CHIP_CLASSES):
    # the server renders badges from it and the page emits it as the JSON
    # blob admin.js derives its maps from.
    env.globals["status_chip_classes"] = STATUS_CHIP_CLASSES  # pyright: ignore[reportArgumentType]  # Why: Jinja2 Environment.globals accepts arbitrary values for template globals.
    env.globals["status_text_classes"] = STATUS_TEXT_CLASSES  # pyright: ignore[reportArgumentType]  # Why: same as above.
    env.globals["badge_chip_base"] = BADGE_CHIP_BASE  # pyright: ignore[reportArgumentType]  # Why: same as above.
    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.
    #
    # The principal capture sits FIRST (where the bare auth dependency used
    # to be): it wraps that same dependency, keeps it running before every
    # page, and stores its return value on request.state.principal so the
    # mutation routes can attribute their audit rows.
    router_dependencies: list[Any] = [
        Depends(_capture_principal_dependency(auth_dependency)),
        Depends(_refresh_clock_offset),
    ]
    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.is_dev_environment
        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

    # The pool and Redis client are resolved per request from app.state,
    # exactly as the admin routes resolve theirs: `taskq ui serve` replaces
    # the pool on a credential rotation, and a router serving from the pool
    # it was constructed with would be serving from a closed one.
    progress_router = _create_progress_router(
        pg_pool,
        redis_client,
        schema=schema,
        auth_dependency=auth_dependency,
        resolve_pg_pool=get_pg_pool,
        resolve_redis_client=get_redis_client,
        session_verifier=session_verifier,
    )
    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,
        actor_fire_policies=actor_fire_policies,
        session_verifier=session_verifier,
    )

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