Settings, Lifespan & Background Tasks

from typing import Annotated

from fastapi import FastAPI
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="BOOKSHELF_")

    app_name: str = "Bookshelf API"
    admin_email: str
    page_size: int = Field(default=20, ge=1, le=100)
    debug: bool = False
    jwt_secret: SecretStr  # no default: startup fails if it is missing


app = FastAPI()
settings = Settings()  # reads os.environ once, at import time


@app.get("/info")
def info() -> dict:
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "page_size": settings.page_size,
        "debug": settings.debug,
        "secret_repr": repr(settings.jwt_secret),
        "secret_len": len(settings.jwt_secret.get_secret_value()),
    }

BOOKSHELF_ADMIN_EMAIL=admin@bookshelf.example \
BOOKSHELF_PAGE_SIZE=50 \
BOOKSHELF_DEBUG=yes \
BOOKSHELF_JWT_SECRET=$(openssl rand -hex 32) \
fastapi run settings_basics.py
curl http://127.0.0.1:8000/info

{"app_name":"Bookshelf API","admin_email":"admin@bookshelf.example","page_size":50,"debug":true,"secret_repr":"SecretStr('**********')","secret_len":64}

os.environ["BOOKSHELF_PAGE_SIZE"] = "500"
del os.environ["BOOKSHELF_JWT_SECRET"]
Settings()

2 validation errors for Settings
page_size
  Input should be less than or equal to 100 [type=less_than_equal, input_value='500', input_type=str]
    For further information visit https://errors.pydantic.dev/2.13/v/less_than_equal
jwt_secret
  Field required [type=missing, input_value={'admin_email': 'admin@bo...: '500', 'debug': 'yes'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing

import secrets, tempfile, pathlib

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

tmp = pathlib.Path(tempfile.mkdtemp())
(tmp / ".env").write_text(
    "BOOKSHELF_APP_NAME=Bookshelf (dev)\n"
    "BOOKSHELF_DATABASE_URL=sqlite:///./dev.db\n"
    f"BOOKSHELF_JWT_SECRET={secrets.token_hex(16)}\n"
    "UNRELATED_TOOL_SETTING=1\n"
)
(tmp / ".env.prod").write_text("BOOKSHELF_APP_NAME=Bookshelf\n")


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="BOOKSHELF_",
        env_file=(tmp / ".env", tmp / ".env.prod"),  # later files override earlier ones
        env_file_encoding="utf-8",
    )
    app_name: str = "Bookshelf API"
    database_url: str = "sqlite:///./bookshelf.db"
    jwt_secret: SecretStr


class LenientSettings(Settings):
    model_config = SettingsConfigDict(extra="ignore")  # merged with the parent's config

>>> Settings()
ValidationError: 1 validation error for Settings
unrelated_tool_setting
  Extra inputs are not permitted [type=extra_forbidden, input_value='1', input_type=str]
>>> LenientSettings()
app_name='Bookshelf' database_url='sqlite:///./dev.db' jwt_secret=SecretStr('**********')

import os, pathlib, tempfile

from pydantic import AliasChoices, Field
from pydantic_settings import BaseSettings, SettingsConfigDict, TomlConfigSettingsSource

tmp = pathlib.Path(tempfile.mkdtemp())
(tmp / "config.toml").write_text('app_name = "From TOML"\npage_size = 33\n')


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="BOOKSHELF_", toml_file=tmp / "config.toml")

    app_name: str = "default"
    page_size: int = 20
    # DATABASE_URL (the name most platforms set) or BOOKSHELF_DATABASE_URL
    database_url: str = Field(
        default="sqlite://",
        validation_alias=AliasChoices("DATABASE_URL", "BOOKSHELF_DATABASE_URL"),
    )

    @classmethod
    def settings_customise_sources(
        cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings
    ):
        return (
            init_settings,
            env_settings,
            dotenv_settings,
            file_secret_settings,
            TomlConfigSettingsSource(settings_cls),  # lowest priority
        )


os.environ["BOOKSHELF_PAGE_SIZE"] = "5"
os.environ["DATABASE_URL"] = "postgresql://db/bookshelf"
print(Settings())

app_name='From TOML' page_size=5 database_url='postgresql://db/bookshelf'

import os

from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class DatabaseSettings(BaseModel):
    url: str = "sqlite:///./bookshelf.db"
    pool_size: int = 5
    echo: bool = False


class RedisSettings(BaseModel):
    host: str = "localhost"
    port: int = 6379
    password: SecretStr | None = None


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="BOOKSHELF_",
        env_nested_delimiter="__",
        env_nested_max_split=1,  # split only on the first delimiter after the prefix
    )
    db: DatabaseSettings = DatabaseSettings()
    redis: RedisSettings = RedisSettings()
    allowed_origins: list[str] = []


os.environ["BOOKSHELF_DB__URL"] = "postgresql+psycopg://bookshelf@db/bookshelf"
os.environ["BOOKSHELF_DB__POOL_SIZE"] = "10"
os.environ["BOOKSHELF_REDIS__HOST"] = "cache.internal"
os.environ["BOOKSHELF_REDIS__PASSWORD"] = os.urandom(6).hex()
os.environ["BOOKSHELF_ALLOWED_ORIGINS"] = '["https://shelf.example", "https://admin.shelf.example"]'
print(Settings())

db=DatabaseSettings(url='postgresql+psycopg://bookshelf@db/bookshelf', pool_size=10, echo=False) redis=RedisSettings(host='cache.internal', port=6379, password=SecretStr('**********')) allowed_origins=['https://shelf.example', 'https://admin.shelf.example']

from typing import Annotated

from pydantic import field_validator
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="BOOKSHELF_")
    allowed_origins: Annotated[list[str], NoDecode] = []

    @field_validator("allowed_origins", mode="before")
    @classmethod
    def split_csv(cls, v):
        if isinstance(v, str):
            return [o.strip() for o in v.split(",") if o.strip()]
        return v

import os, pathlib, tempfile

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

secrets = pathlib.Path(tempfile.mkdtemp())  # stands in for /run/secrets
(secrets / "jwt_secret").write_text(os.urandom(16).hex() + "\n")
(secrets / "db_password").write_text(os.urandom(8).hex())


class Settings(BaseSettings):
    model_config = SettingsConfigDict(secrets_dir=secrets)  # in a container: "/run/secrets"
    jwt_secret: SecretStr
    db_password: SecretStr
    region: str = "eu-west-1"


s = Settings()
print(repr(s.jwt_secret), len(s.jwt_secret.get_secret_value()), repr(s.db_password))
os.environ["JWT_SECRET"] = os.urandom(4).hex()  # a variable with the same name wins
print(len(Settings().jwt_secret.get_secret_value()))

SecretStr('**********') 32 SecretStr('**********')
8

from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, NestedSecretsSettingsSource, SettingsConfigDict


class Db(BaseModel):
    user: str = "bookshelf"
    password: SecretStr


class Settings(BaseSettings):
    model_config = SettingsConfigDict(secrets_dir="secrets", secrets_nested_subdir=True)
    db: Db

    @classmethod
    def settings_customise_sources(
        cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings
    ):
        return (init_settings, env_settings, dotenv_settings, NestedSecretsSettingsSource(file_secret_settings))

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated, TypedDict

import httpx
from fastapi import Depends, FastAPI, Request

logger = logging.getLogger("bookshelf")
logging.basicConfig(level=logging.INFO, format="%(message)s")


class State(TypedDict):
    http: httpx.AsyncClient
    isbn_cache: dict[str, str]


def fake_isbn_service(request: httpx.Request) -> httpx.Response:
    isbn = request.url.path.rsplit("/", 1)[-1]
    return httpx.Response(200, json={"isbn": isbn, "title": "Dune"})


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    logger.info("startup: creating the shared HTTP client")
    async with httpx.AsyncClient(
        base_url="https://isbn.example",
        transport=httpx.MockTransport(fake_isbn_service),  # replace in production
        timeout=httpx.Timeout(5.0),
    ) as http:
        yield {"http": http, "isbn_cache": {}}  # becomes request.state
    logger.info("shutdown: HTTP client closed")  # reached after the yield


app = FastAPI(title="Bookshelf API", lifespan=lifespan)


def get_http(request: Request) -> httpx.AsyncClient:
    return request.state.http


HttpDep = Annotated[httpx.AsyncClient, Depends(get_http)]


@app.get("/lookup/{isbn}")
async def lookup(isbn: str, http: HttpDep, request: Request) -> dict:
    cache: dict[str, str] = request.state.isbn_cache
    if isbn not in cache:
        response = await http.get(f"/isbn/{isbn}")
        cache[isbn] = response.json()["title"]
    return {"isbn": isbn, "title": cache[isbn], "cached_items": len(cache)}

from fastapi.testclient import TestClient

with TestClient(app) as client:  # `with` is what runs startup and shutdown
    print(client.get("/lookup/978-0441172719").json())
    print(client.get("/lookup/978-0441172719").json())

startup: creating the shared HTTP client
{'isbn': '978-0441172719', 'title': 'Dune', 'cached_items': 1}
{'isbn': '978-0441172719', 'title': 'Dune', 'cached_items': 1}
shutdown: HTTP client closed

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from functools import lru_cache
from typing import Annotated

from fastapi import Depends, FastAPI, Request
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

logger = logging.getLogger("bookshelf")


class DatabaseSettings(BaseModel):
    url: str = "sqlite:///./bookshelf.db"
    pool_size: int = Field(default=5, ge=1)


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="BOOKSHELF_",
        env_file=".env",
        env_nested_delimiter="__",
        extra="ignore",
    )
    app_name: str = "Bookshelf API"
    db: DatabaseSettings = DatabaseSettings()
    jwt_secret: SecretStr  # required, no default


@lru_cache
def get_settings() -> Settings:
    return Settings()


SettingsDep = Annotated[Settings, Depends(get_settings)]


class Pool:
    """Stands in for a SQLAlchemy engine or an asyncpg/redis pool."""

    def __init__(self, url: str, size: int) -> None:
        self.url, self.size, self.closed = url, size, False

    def close(self) -> None:
        self.closed = True
        logger.info("pool closed")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict]:
    # a plain call: dependency_overrides do NOT apply here, so look the override up explicitly
    settings = app.dependency_overrides.get(get_settings, get_settings)()
    pool = Pool(settings.db.url, settings.db.pool_size)
    logger.info("pool opened: %s (size %d)", pool.url, pool.size)
    try:
        yield {"pool": pool}
    finally:
        pool.close()


app = FastAPI(title="Bookshelf API", lifespan=lifespan)


@app.get("/status")
def status(request: Request, settings: SettingsDep) -> dict:
    pool: Pool = request.state.pool
    return {"app": settings.app_name, "pool_url": pool.url, "pool_size": pool.size, "closed": pool.closed}

BOOKSHELF_JWT_SECRET=$(openssl rand -hex 32) BOOKSHELF_DB__POOL_SIZE=8 fastapi run full.py
curl http://127.0.0.1:8000/status

{"app":"Bookshelf (dev)","pool_url":"sqlite:///./dev.db","pool_size":8,"closed":false}

pydantic_core._pydantic_core.ValidationError: 1 validation error for Settings
jwt_secret
  Field required [type=missing, input_value={'app_name': 'Bookshelf (...: 'sqlite:///./dev.db'}}, input_type=dict]
ERROR:    Application startup failed. Exiting.

import secrets

import pytest
from fastapi.testclient import TestClient

from full import Settings, app, get_settings


def test_status_with_override():
    app.dependency_overrides[get_settings] = lambda: Settings(
        _env_file=None,                      # ignore any developer .env
        jwt_secret=secrets.token_hex(16),    # random per run, never a real key
        db={"url": "sqlite:///:memory:", "pool_size": 1},
    )
    try:
        with TestClient(app) as client:       # `with` runs the lifespan
            assert client.get("/status").json() == {
                "app": "Bookshelf API",
                "pool_url": "sqlite:///:memory:",
                "pool_size": 1,
                "closed": False,
            }
    finally:
        app.dependency_overrides.clear()


def test_fail_fast_without_secret(monkeypatch):
    monkeypatch.delenv("BOOKSHELF_JWT_SECRET", raising=False)
    monkeypatch.chdir("/")                 # no .env here
    get_settings.cache_clear()
    with pytest.raises(Exception) as exc:
        with TestClient(app):
            pass
    assert "jwt_secret" in str(exc.value)
    get_settings.cache_clear()

uvicorn serve_l:app --workers 2

INFO:     Started server process [13282]
INFO:     Started server process [13283]
INFO:     Waiting for application startup.
INFO:     Waiting for application startup.
INFO:     startup in pid 13282: opening pool
INFO:     startup in pid 13283: opening pool
INFO:     Application startup complete.
INFO:     Application startup complete.
...
INFO:     Shutting down
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Waiting for application shutdown.
INFO:     shutdown in pid 13282: closing pool
INFO:     shutdown in pid 13283: closing pool
INFO:     Application shutdown complete.
INFO:     Application shutdown complete.

from contextlib import asynccontextmanager

from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient

events: list[str] = []


def make_lifespan(name: str):
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        events.append(f"{name} startup")
        yield {"owner": name}
        events.append(f"{name} shutdown")
    return lifespan


# 1. a mounted sub-application with its own lifespan
admin = FastAPI(title="Admin", lifespan=make_lifespan("admin"))


@admin.get("/ping")
def admin_ping():
    return {"ok": True}


main = FastAPI(title="Bookshelf API", lifespan=make_lifespan("main"))
main.mount("/admin", admin)

with TestClient(main) as c:
    c.get("/admin/ping")
print("mounted:", events)

# 2. a router with a lifespan, included in the main app
events.clear()
router = APIRouter(prefix="/reviews", lifespan=make_lifespan("reviews-router"))


@router.get("/")
def reviews():
    return []


main2 = FastAPI(lifespan=make_lifespan("main"))
main2.include_router(router)
with TestClient(main2) as c:
    c.get("/reviews/")
print("included router:", events)

# 3. the parent runs the sub-application's lifespan explicitly
events.clear()


@asynccontextmanager
async def parent_lifespan(app: FastAPI):
    events.append("main startup")
    async with admin.router.lifespan_context(admin):
        yield
    events.append("main shutdown")


main3 = FastAPI(lifespan=parent_lifespan)
main3.mount("/admin", admin)
with TestClient(main3) as c:
    c.get("/admin/ping")
print("explicit:", events)

mounted: ['main startup', 'main shutdown']
included router: ['main startup', 'reviews-router startup', 'reviews-router shutdown', 'main shutdown']
explicit: ['main startup', 'admin startup', 'admin shutdown', 'main shutdown']

import warnings

from fastapi import FastAPI
from fastapi.testclient import TestClient

log: list[str] = []
old = FastAPI()


@old.on_event("startup")  # deprecated: emits DeprecationWarning
async def startup() -> None:
    log.append("startup")


@old.on_event("shutdown")
def shutdown() -> None:
    log.append("shutdown")


with TestClient(old):
    pass
print(log)  # ['startup', 'shutdown']

import asyncio
import logging
from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel

logger = logging.getLogger("bookshelf")
events: list[str] = []


def record(msg: str) -> None:
    try:
        asyncio.get_running_loop()
        where = "event loop"
    except RuntimeError:
        where = "threadpool"
    events.append(f"{msg} [{where}]")


class ReviewIn(BaseModel):
    reader: str
    stars: int


def send_thanks_email(address: str, *, subject: str) -> None:   # def: runs in the threadpool
    record(f"email to {address}: {subject}")


async def refresh_rating_cache(isbn: str) -> None:               # async def: runs on the event loop
    await asyncio.sleep(0)
    record(f"rating cache refreshed for {isbn}")


def audit(background_tasks: BackgroundTasks, x_request_id: Annotated[str, Header()] = "n/a") -> str:
    background_tasks.add_task(record, f"audit request {x_request_id}")   # added from a dependency
    return x_request_id


app = FastAPI(title="Bookshelf API")


@app.post("/books/{isbn}/reviews", status_code=201)
async def add_review(
    isbn: str,
    review: ReviewIn,
    background_tasks: BackgroundTasks,
    request_id: Annotated[str, Depends(audit)],
) -> ReviewIn:
    if review.stars == 0:
        raise HTTPException(status_code=422, detail="stars must be 1..5")
    background_tasks.add_task(send_thanks_email, f"{review.reader}@example.com", subject="Thanks!")
    background_tasks.add_task(refresh_rating_cache, isbn)
    record("path function done")
    return review

curl -i -X POST http://127.0.0.1:8000/books/978-0441172719/reviews \
  -H 'Content-Type: application/json' -H 'X-Request-Id: req-1' \
  -d '{"reader":"ana","stars":5}'

{"reader":"ana","stars":5}

path function done [event loop]
audit request req-1 [threadpool]
email to ana@example.com: Thanks! [threadpool]
rating cache refreshed for 978-0441172719 [event loop]

from fastapi import BackgroundTasks, FastAPI
from fastapi.testclient import TestClient

ran: list[str] = []
app = FastAPI()


def ok(n): ran.append(f"task {n}")
def boom(): raise RuntimeError("smtp server is down")


@app.post("/failing")
def failing(tasks: BackgroundTasks):
    tasks.add_task(ok, 1); tasks.add_task(boom); tasks.add_task(ok, 3)
    return {"accepted": True}


with TestClient(app, raise_server_exceptions=False) as c:
    r = c.post("/failing")
    print(r.status_code, r.json(), ran)

200 {'accepted': True} ['task 1']

import asyncio
import logging
import time

from fastapi import BackgroundTasks, FastAPI

log = logging.getLogger("uvicorn.error")
app = FastAPI(title="Bookshelf API")


def build_report_sync(isbn: str) -> None:      # def: threadpool, does not block the loop
    time.sleep(2)
    log.info("sync report for %s finished", isbn)


async def build_report_blocking(isbn: str) -> None:  # async def with a blocking call: freezes the loop
    time.sleep(2)
    log.info("blocking async report for %s finished", isbn)


@app.post("/reports/sync/{isbn}", status_code=202)
def report_sync(isbn: str, tasks: BackgroundTasks) -> dict:
    tasks.add_task(build_report_sync, isbn)
    return {"status": "accepted"}


@app.post("/reports/blocking/{isbn}", status_code=202)
def report_blocking(isbn: str, tasks: BackgroundTasks) -> dict:
    tasks.add_task(build_report_blocking, isbn)
    return {"status": "accepted"}


@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

curl -s -o /dev/null -w "POST %{time_total}s\n" -X POST http://127.0.0.1:8124/reports/sync/978-0441172719
curl -s -o /dev/null -w "health during task %{time_total}s\n" http://127.0.0.1:8124/health
curl -s -o /dev/null -w "POST %{time_total}s\n" -X POST http://127.0.0.1:8124/reports/blocking/978-0441172719
curl -s -o /dev/null -w "health during task %{time_total}s\n" http://127.0.0.1:8124/health

POST 0.034281s
health during task 0.001130s
POST 0.000871s
health during task 1.692800s