Metadata-Version: 2.4
Name: fastapi-typed-state
Version: 0.2.0
Summary: Type-safe application state for FastAPI
Keywords: fastapi,state,typing
Author: Florian Daude
Author-email: Florian Daude <floriandaude@hotmail.fr>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Dist: fastapi>=0.115
Requires-Python: >=3.12
Project-URL: Homepage, https://gitlab.com/daude_f/fastapi-typed-state
Project-URL: Repository, https://gitlab.com/daude_f/fastapi-typed-state
Project-URL: Issues, https://gitlab.com/daude_f/fastapi-typed-state/-/issues
Description-Content-Type: text/markdown

# 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

```bash
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:

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

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

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

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

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