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
    ),
) -> 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).

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),
) -> 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).
    """
    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()
    _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,
    ) -> EventSourceResponse:
        """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"},
            )

        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),
            )
            with contextlib.suppress(Exception):
                await pubsub.aclose()
            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"},
            )

        # 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)
            with contextlib.suppress(Exception):
                await pubsub.aclose()
            raise

        if row is None:
            with contextlib.suppress(Exception):
                await pubsub.unsubscribe(channel)
            with contextlib.suppress(Exception):
                await pubsub.aclose()
            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,
                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,
)

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.

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

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_schema

get_schema(request: Request) -> str

Dependency: yields the schema name from app.state.

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

def get_schema(request: Request) -> str:
    """Dependency: yields the schema name from ``app.state``."""
    s: str = request.app.state.schema
    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

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,
) -> 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.

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,
) -> 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.
    """
    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

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

    router = APIRouter(**router_kwargs)

    if auth_dependency is None and settings.environment not in {"dev", "development"}:
        if 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)"
            )
        logger.warning(
            "admin-ui-no-auth",
            environment=settings.environment,
        )

    @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,
    )

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