fastapi-typed-state

Type-safe application state for FastAPI lifespans, dependencies, requests, and WebSockets.

The library stores one application-scoped object in a private app.state slot. Endpoints can retrieve that object through a typed dependency or manually from any Starlette HTTPConnection.

Installation

pip install fastapi-typed-state

Python 3.12 or newer is required.

Usage

The following example loads non-secret configuration during startup, fetches database credentials from AWS Secrets Manager, and keeps a PostgreSQL pool and S3 client open for the application lifespan.

The example uses additional application dependencies that are not required by fastapi-typed-state:

pip install asyncpg aiobotocore types-aiobotocore-s3 types-aiobotocore-secrets-manager

config.toml contains identifiers and deployment configuration, but no credentials:

aws_region = 'eu-west-1'
s3_bucket = 'reports-production'
database_secret_id = 'production/reports/database'

import tomllib
from collections.abc import AsyncGenerator
from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass
from pathlib import Path

import asyncpg
from aiobotocore.session import get_session
from fastapi import FastAPI, Request
from pydantic import BaseModel, ConfigDict, PostgresDsn
from types_aiobotocore_s3.client import S3Client

import fastapi_typed_state


class ApplicationConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    aws_region: str
    s3_bucket: str
    database_secret_id: str


class DatabaseSecret(BaseModel):
    model_config = ConfigDict(frozen=True)

    dsn: PostgresDsn


@dataclass(frozen=True, slots=True)
class ApplicationState:
    config: ApplicationConfig
    database: asyncpg.Pool
    s3: S3Client


AppStateDep = fastapi_typed_state.Extracted[ApplicationState]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    with Path('config.toml').open('rb') as config_file:
        config = ApplicationConfig.model_validate(tomllib.load(config_file))

    aws = get_session()
    async with aws.create_client('secretsmanager', region_name=config.aws_region) as secret_manager:
        secret_response = await secret_manager.get_secret_value(SecretId=config.database_secret_id)

    secret_payload = secret_response.get('SecretString')
    if secret_payload is None:
        msg = 'Database secret does not contain SecretString'
        raise RuntimeError(msg)
    database_secret = DatabaseSecret.model_validate_json(secret_payload)

    async with AsyncExitStack() as resources:
        database = await asyncpg.create_pool(dsn=str(database_secret.dsn))
        resources.push_async_callback(database.close)
        s3 = await resources.enter_async_context(
            aws.create_client('s3', region_name=config.aws_region)
        )
        state = ApplicationState(config=config, database=database, s3=s3)

        async with fastapi_typed_state.context(app, state):
            yield


app = FastAPI(lifespan=lifespan)


@app.get('/ready')
async def ready(state: AppStateDep) -> dict[str, str]:
    database_value = await state.database.fetchval('SELECT 1')
    await state.s3.head_bucket(Bucket=state.config.s3_bucket)
    return {'database': 'ready' if database_value == 1 else 'not ready', 's3': 'ready'}


@app.get('/configuration')
async def configuration(request: Request) -> dict[str, str]:
    state = fastapi_typed_state.extract(request, ApplicationState)
    return {'aws_region': state.config.aws_region, 's3_bucket': state.config.s3_bucket}

The Secrets Manager client closes immediately after startup retrieves and validates the secret. On shutdown, context() first removes the application state, then AsyncExitStack closes the S3 client and database pool.

Importing the module keeps the intentionally concise API names explicit at each call site. The direct assignment to AppStateDep preserves FastAPI's dependency metadata at runtime and gives endpoints a short annotation. For Pyright, both AppStateDep and fastapi_typed_state.Extracted[ApplicationState] are statically the same type as ApplicationState.

At runtime, Extracted[...] retrieves the lifespan object and verifies it with isinstance. extract(connection, ApplicationState) performs the same check and has the concrete return type ApplicationState. It accepts Starlette's HTTPConnection, so it works with both FastAPI Request and WebSocket objects. Subclasses satisfy an expected base class through normal isinstance semantics.

Lifecycle

The state is available after entering context and is removed when the context exits, including exceptional exits. The supplied object is not opened or closed by this library; compose its own context manager outside context when it manages resources.

Only one fastapi-typed-state object can be active for an application. Starting a second context on the same application raises StateAlreadyInitializedError without replacing the active object. Different FastAPI applications have independent state.

Errors

Manual retrieval raises public library exceptions:

• StateNotInitializedError when the application lifespan is not active.
• StateTypeMismatchError when the object does not match the expected class.
• StateAlreadyInitializedError when a second context targets the same application.

All three inherit from TypedStateError.

Dependency retrieval converts missing state and type mismatches into HTTP 500 responses with stable details. Responses do not contain the stored object or its representation.

Both Extracted[...] and extract(..., expected_type) require a concrete runtime class. Any, unions, and parameterized generics such as list[str] are not supported because they cannot be checked with ordinary isinstance semantics.

Development

uv sync
uv run pyright
uv run ruff format --check .
uv run ruff check .
uv run pytest
uv build

Pyright runs in strict mode using its Node.js extra. Ruff checks all rules except return annotations, docstrings, copyright, security, and lazy-import rules; formatting uses spaces, LF line endings, a 100-character line length, single quotes, and no magic trailing comma.