FastAPI Best Practices

https://github.com/zhanymkanov/fastapi-best-practices
Admin
FastAPI Best Practices and Conventions we used at our startup
Tokens:12,847
Snippets:155
Trust Score:8.8
Update:5 months ago
Show doc for...
Tokens:
Raw
### FastAPI Project Structure Example

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Illustrates a scalable and evolvable project structure for a FastAPI monolith, organizing code by domain within the 'src' directory. Each domain package includes its own routers, schemas, models, services, dependencies, constants, configurations, utilities, and exceptions.

```tree
fastapi-project
├── alembic/
├── src
│   ├── auth
│   │   ├── router.py
│   │   ├── schemas.py  # pydantic models
│   │   ├── models.py  # db models
│   │   ├── dependencies.py
│   │   ├── config.py  # local configs
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── aws
│   │   ├── client.py  # client model for external service communication
│   │   ├── schemas.py
│   │   ├── config.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   └── utils.py
│   ├── posts
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── dependencies.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── config.py  # global configs
│   ├── models.py  # global models
│   ├── exceptions.py  # global exceptions
│   ├── pagination.py  # global module e.g. pagination
│   ├── database.py  # db connection related stuff
│   └── main.py
├── tests/
│   ├── auth
│   ├── aws
│   └── posts
├── templates/
│   └── index.html
├── requirements
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
```

--------------------------------

### Documenting API Endpoints with Custom Responses

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

This example demonstrates how to thoroughly document a FastAPI endpoint, including specifying response models, status codes, descriptions, and handling multiple possible responses.

```APIDOC
## Documenting API Endpoints with Custom Responses

### Description
This example demonstrates how to thoroughly document a FastAPI endpoint, including specifying response models, status codes, descriptions, and handling multiple possible responses.

### Method
POST

### Endpoint
/endpoints

### Parameters
#### Request Body
(Not specified in the source, but implied by the route definition)

### Response
#### Success Response (200)
- **model**: OkResponse - custom pydantic model for 200 response
- **description**: Ok Response

#### Success Response (201)
- **model**: CreatedResponse - custom pydantic model for 201 response
- **description**: Creates something from user request

#### Success Response (202)
- **model**: AcceptedResponse - custom pydantic model for 202 response
- **description**: Accepts request and handles it later

### Code
```python
from fastapi import APIRouter, status

router = APIRouter()

@router.post(
    "/endpoints",
    response_model=DefaultResponseModel,  # default response pydantic model 
    status_code=status.HTTP_201_CREATED,  # default status code
    description="Description of the well documented endpoint",
    tags=["Endpoint Category"],
    summary="Summary of the Endpoint",
    responses={
        status.HTTP_200_OK: {
            "model": OkResponse, # custom pydantic model for 200 response
            "description": "Ok Response",
        },
        status.HTTP_201_CREATED: {
            "model": CreatedResponse,  # custom pydantic model for 201 response
            "description": "Creates something from user request",
        },
        status.HTTP_202_ACCEPTED: {
            "model": AcceptedResponse,  # custom pydantic model for 202 response
            "description": "Accepts request and handles it later",
        },
    },
)
async def documented_route():
    pass
```
```

--------------------------------

### Auth Configuration with Pydantic-Settings

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Define authentication settings using `pydantic-settings.BaseSettings`. This example configures JWT parameters and refresh token expiration, ignoring extra environment variables.

```python
# src/auth/config.py
from datetime import timedelta
from pydantic_settings import BaseSettings, SettingsConfigDict


class AuthConfig(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="AUTH_", env_file=".env", extra="ignore")

    JWT_ALG: str
    JWT_SECRET: str
    JWT_EXP_MINUTES: int = 5
    REFRESH_TOKEN_KEY: str
    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
    SECURE_COOKIES: bool = True


auth_settings = AuthConfig()
```

--------------------------------

### Validating Dependencies in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Implement dependency functions that not only inject but also validate data. This example fetches a post by ID and raises `PostNotFound` if it doesn't exist.

```python
async def valid_post_id(post_id: UUID4) -> dict:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()
    return post
```

--------------------------------

### Advanced Dependency Chaining in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Illustrates a more complex dependency chain involving `valid_post_id`, `parse_jwt_data`, `valid_owned_post`, and `valid_active_creator`. This setup ensures that a post is valid, owned by the user, and that the user is active before processing the request.

```python
from fastapi import BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
import jwt  # PyJWT
from jwt.exceptions import InvalidTokenError

async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except InvalidTokenError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: Mapping = Depends(valid_post_id), 
    token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post


async def valid_active_creator(
    token_data: dict = Depends(parse_jwt_data),
):
    user = await users_service.get_by_id(token_data["user_id"])
    if not user["is_active"]:
        raise UserIsBanned()
    
    if not user["is_creator"]:
       raise UserNotCreator()
    
    return user
        

# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(
    worker: BackgroundTasks,
    post: Mapping = Depends(valid_owned_post),
    user: Mapping = Depends(valid_active_creator),
):
    """Get post that belong the active user."""
    worker.add_task(notifications_service.send_email, user["id"])
    return post

```

--------------------------------

### FastAPI Response Serialization Example

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates how FastAPI serializes responses. FastAPI converts Pydantic objects to dicts using jsonable_encoder, then validates against response_model, and finally serializes to JSON. This results in the Pydantic model being created twice.

```python
from fastapi import FastAPI
from pydantic import BaseModel, model_validator

app = FastAPI()


class ProfileResponse(BaseModel):
    @model_validator(mode="after")
    def debug_usage(self):
        print("created pydantic model")

        return self


@app.get("/", response_model=ProfileResponse)
async def root():
    return ProfileResponse()
```

--------------------------------

### Custom Field Serialization with Pydantic v2

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use `@field_serializer` for custom per-field serialization rules in Pydantic models. This example serializes datetime objects to ISO 8601 format with timezone information.

```python
from datetime import datetime
from zoneinfo import ZoneInfo
from pydantic import BaseModel, ConfigDict, field_serializer


class CustomModel(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    @field_serializer("*", when_used="json", check_fields=False)
    def _serialize_datetimes(self, value):
        if isinstance(value, datetime):
            if value.tzinfo is None:
                value = value.replace(tzinfo=ZoneInfo("UTC"))
            return value.strftime("%Y-%m-%dT%H:%M:%S%z")
        return value
```

--------------------------------

### Configure Alembic async template and file naming

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Initialize Alembic with the async template for asynchronous database operations and configure the file_template in alembic.ini for descriptive migration filenames.

```ini
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s
```

--------------------------------

### Set up Async Test Client with httpx

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Configures an asynchronous test client using httpx and ASGITransport for integration testing FastAPI applications. This avoids event loop issues.

```python
from typing import AsyncGenerator

import pytest
from httpx import AsyncClient, ASGITransport

from src.main import app  # inited FastAPI app


@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_create_post(client: AsyncClient):
    resp = await client.post("/posts")

    assert resp.status_code == 201
```

--------------------------------

### FastAPI Quick Reference: Linting and Formatting

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use `ruff check --fix` for linting and fixing code, and `ruff format` for formatting code.

```python
ruff check --fix
```

```python
ruff format
```

--------------------------------

### Create FastAPI Router for Posts

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Sets up a FastAPI router to handle requests for creator posts. It depends on a `valid_creator_id` dependency to ensure the creator ID is valid before fetching posts.

```python
from fastapi import APIRouter, Depends

router = APIRouter()


@router.get("/creators/{creator_id}/posts", response_model=list[Post])
async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)):
   posts = await service.get_posts(creator["id"])

   return posts
```

--------------------------------

### FastAPI Quick Reference: Non-blocking I/O

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use `async def` routes with `await` for non-blocking I/O operations.

```python
async def route with await
```

--------------------------------

### FastAPI Async vs. Sync Route Handling

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates three ways to handle blocking I/O in FastAPI: a terrible async approach that blocks the event loop, a good sync approach that uses a threadpool, and a perfect async approach using asyncio.sleep.

```python
import asyncio
import time

from fastapi import APIRouter


router = APIRouter()


@router.get("/terrible-ping")
async def terrible_ping():
    time.sleep(10) # I/O blocking operation for 10 seconds, the whole process will be blocked
    
    return {"pong": True}

@router.get("/good-ping")
def good_ping():
    time.sleep(10) # I/O blocking operation for 10 seconds, but in a separate thread for the whole `good_ping` route

    return {"pong": True}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10) # non-blocking I/O operation

    return {"pong": True}
```

--------------------------------

### Importing Modules in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates how to import modules from different parts of the project using explicit module names. This ensures clarity and avoids potential naming conflicts when dependencies exist across various domain packages.

```python
from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode  # in case we have Standard ErrorCode in constants module of each package
```

--------------------------------

### Use httpx for Async Testing

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

The `async_asgi_testclient` is unmaintained. Use `httpx.AsyncClient` with `ASGITransport` for testing ASGI applications.

```python
httpx.AsyncClient
```

--------------------------------

### Set up AsyncClient for FastAPI testing with httpx

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

This fixture provides an AsyncClient for testing FastAPI applications using httpx and ASGITransport. It's the recommended approach for asynchronous testing.

```python
import pytest
from httpx import AsyncClient, ASGITransport

from src.main import app


@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_create_post(client: AsyncClient):
    resp = await client.post("/posts", json={"title": "hi"})
    assert resp.status_code == 201
```

--------------------------------

### FastAPI Quick Reference: HTTP Test Client

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use `httpx.AsyncClient` combined with `ASGITransport` as an HTTP test client for your FastAPI application.

```python
httpx.AsyncClient + ASGITransport
```

--------------------------------

### FastAPI Quick Reference: Async Database

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use SQLAlchemy 2.0's async capabilities with `AsyncSession` for asynchronous database interactions.

```python
SQLAlchemy 2.0 async (AsyncSession)
```

--------------------------------

### FastAPI Quick Reference: Fire-and-Forget Tasks

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

For simple fire-and-forget tasks, use FastAPI's `BackgroundTasks`.

```python
BackgroundTasks
```

--------------------------------

### Use Real Databases for Integration Tests

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Mocking the database in integration tests can lead to divergence from production behavior. Use a real database (e.g., via testcontainers or an ephemeral schema) and `dependency_overrides` for external services like authentication.

```python
app.dependency_overrides[dep] = fake
```

--------------------------------

### FastAPI Quick Reference: Blocking I/O (Sync)

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

For blocking I/O without an async client, use a synchronous `def` route, which runs in a threadpool.

```python
def route (sync, runs in threadpool)
```

--------------------------------

### Running Sync SDK in Thread Pool with FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Illustrates how to run synchronous SDK calls within an asynchronous FastAPI application using `run_in_threadpool` from Starlette. This is essential for libraries that do not support async operations.

```python
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from my_sync_library import SyncAPIClient 

app = FastAPI()


@app.get("/")
async def call_my_sync_library():
    my_data = await service.get_my_data()

    client = SyncAPIClient()
    await run_in_threadpool(client.make_request, data=my_data)
```

--------------------------------

### Handling Blocking I/O in FastAPI Async Routes

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Demonstrates how to correctly handle blocking I/O operations within FastAPI routes. Use `async def` for non-blocking I/O and `def` for blocking I/O to run in a threadpool. For sync libraries within async routes, use `run_in_threadpool`.

```python
# DON'T — blocking call inside async route freezes the entire event loop
@router.get("/bad")
async def bad():
    time.sleep(10)            # blocks every request on this worker
    return {"ok": True}
```

```python
# DO — sync route lets FastAPI run it in a threadpool
@router.get("/sync-ok")
def sync_ok():
    time.sleep(10)            # blocks one threadpool worker, not the loop
    return {"ok": True}
```

```python
# DO — async route with awaitable sleep
@router.get("/async-ok")
async def async_ok():
    await asyncio.sleep(10)   # yields control, loop keeps serving requests
    return {"ok": True}
```

```python
# DO — async route that has to call a sync library
from fastapi.concurrency import run_in_threadpool

@router.get("/wrap")
async def wrap():
    result = await run_in_threadpool(legacy_sync_client.fetch, "id")
    return result
```

--------------------------------

### FastAPI Quick Reference: Request Validation Against DB

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Implement request validation against a database using a dependency that loads, validates, and returns the required data.

```python
Dependency that loads + validates + returns
```

--------------------------------

### Fetch Posts with Creator Information

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Retrieves posts for a given creator, joining with profile information to include creator details. Uses SQLAlchemy for database interaction.

```python
from typing import Any

from pydantic import UUID4
from sqlalchemy import desc, func, select, text
from sqlalchemy.sql.functions import coalesce

from src.database import database, posts, profiles, post_review, products

async def get_posts(
    creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[dict[str, Any]]: 
    select_query = (
        select(
            (
                posts.c.id,
                posts.c.slug,
                posts.c.title,
                func.json_build_object(
                   text("'id', profiles.id")
                   text("'first_name', profiles.first_name")
                   text("'last_name', profiles.last_name")
                   text("'username', profiles.username")
                ).label("creator"),
            )
        )
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
        .where(posts.c.owner_id == creator_id)
        .limit(limit)
        .offset(offset)
        .group_by(
            posts.c.id,
            posts.c.type,
            posts.c.slug,
            posts.c.title,
            profiles.c.id,
            profiles.c.first_name,
            profiles.c.last_name,
            profiles.c.username,
            profiles.c.avatar,
        )
        .order_by(
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
        )
    )
    
    return await database.fetch_all(select_query)
```

--------------------------------

### Isolate Configuration per Domain

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Reusing a single `BaseSettings` instance for the entire application makes it hard to reason about and leads to unnecessary variable reads. Create a separate `BaseSettings` subclass for each domain.

```python
One BaseSettings subclass per domain
```

--------------------------------

### Define Dependencies for RESTful Routing

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Shows how to create dependencies for validating resource IDs, such as `valid_profile_id`. This pattern supports RESTful API design by ensuring resources exist before further processing.

```python
# src.profiles.dependencies
async def valid_profile_id(profile_id: UUID4) -> Mapping:
    profile = await service.get_by_id(profile_id)
    if not profile:
        raise ProfileNotFound()

    return profile
```

--------------------------------

### FastAPI Quick Reference: Sync Library in Async Route

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

To use a synchronous library within an async route, wrap the call with `await run_in_threadpool(fn, *args)`.

```python
await run_in_threadpool(fn, *args)
```

--------------------------------

### Decouple Pydantic BaseSettings into Modules

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Organize application settings by creating separate Pydantic BaseSettings classes for different modules or domains, such as authentication and general configuration. This prevents a monolithic settings file and improves maintainability.

```python
# src.auth.config
from datetime import timedelta

from pydantic_settings import BaseSettings


class AuthConfig(BaseSettings):
    JWT_ALG: str
    JWT_SECRET: str
    JWT_EXP: int = 5  # minutes

    REFRESH_TOKEN_KEY: str
    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)

    SECURE_COOKIES: bool = True


auth_settings = AuthConfig()


# src.config
from pydantic import PostgresDsn, RedisDsn
from pydantic_settings import BaseSettings

from src.constants import Environment


class Config(BaseSettings):
    DATABASE_URL: PostgresDsn
    REDIS_URL: RedisDsn

    SITE_DOMAIN: str = "myapp.com"

    ENVIRONMENT: Environment = Environment.PRODUCTION

    SENTRY_DSN: str | None = None

    CORS_ORIGINS: list[str]
    CORS_ORIGINS_REGEX: str | None = None
    CORS_HEADERS: list[str]

    APP_VERSION: str = "1.0"


settings = Config()
```

--------------------------------

### Async SQLAlchemy 2.0 Database Session Management

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Set up an asynchronous database engine and session factory using SQLAlchemy 2.0. The `get_db` dependency provides an `AsyncSession` for request-scoped database operations.

```python
# src/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def get_db() -> AsyncSession:
    async with SessionFactory() as session:
        yield session
```

--------------------------------

### FastAPI Quick Reference: Reusing Validation

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Reuse validation logic across multiple routes by chaining dependencies.

```python
Chain dependencies
```

--------------------------------

### Using FastAPI BackgroundTasks for Short In-Process Tasks

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Shows how to use FastAPI's `BackgroundTasks` for short, in-process tasks that should run after the response is sent. This is suitable for tasks like sending emails or logging, but not for long-running or critical operations.

```python
from fastapi import BackgroundTasks


@router.post("/signup")
async def signup(data: SignupIn, bg: BackgroundTasks):
    user = await service.create_user(data)
    bg.add_task(send_welcome_email, user.email)  # fire-and-forget, in-process
    return user
```

--------------------------------

### Chain Dependencies for Creator Validation

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates chaining dependencies to enforce creator-specific logic. `valid_creator_id` depends on `valid_profile_id`, ensuring that a profile is first validated and then checked for creator status.

```python
# src.creators.dependencies
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
    if not profile["is_creator"]:
       raise ProfileNotCreator()

    return profile
```

--------------------------------

### Modern Dependency Injection Style

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

The default-arg form for dependencies is legacy and can have gotchas. Use `Annotated[T, Depends(...)]` for a more robust and modern approach.

```python
user: Annotated[User, Depends(...)]
```

--------------------------------

### FastAPI Quick Reference: Modern Dependency Injection

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Inject dependencies using the modern `Annotated[T, Depends(...)]` syntax.

```python
Annotated[T, Depends(...)]
```

--------------------------------

### Lint and format Python code with Ruff

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use Ruff for linting and formatting Python code. It replaces tools like black, isort, and flake8. Add these commands to pre-commit hooks or CI pipelines.

```shell
ruff check --fix src
ruff format src
```

--------------------------------

### FastAPI Quick Reference: JWT Decode

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Decode JWT tokens using the `PyJWT` library (`import jwt`).

```python
PyJWT (import jwt)
```

--------------------------------

### FastAPI Quick Reference: CPU-Intensive Work

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Offload CPU-intensive work to separate worker processes using task queues like Celery, Arq, or RQ.

```python
Celery / Arq / RQ worker process
```

--------------------------------

### Run Ruff for Linting and Formatting

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Executes Ruff commands to check for linting errors and automatically fix them, as well as format the code. This ensures code quality and consistency.

```shell
#!/bin/sh -e
set -x

ruff check --fix src
ruff format src
```

--------------------------------

### Chaining Dependencies for Reuse in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Combine multiple dependencies to create reusable, composable logic. This dependency ensures the post is valid and owned by the authenticated user.

```python
async def valid_owned_post(
    post: Annotated[dict, Depends(valid_post_id)],
    token_data: Annotated[dict, Depends(parse_jwt_data)],
) -> dict:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()
    return post
```

--------------------------------

### Decouple Imports with Aliases

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Importing across domains using deep paths (`from src.auth.service.user import ...`) creates tight coupling and makes refactoring difficult. Use module aliases for cleaner imports (`from src.auth import service as auth_service`).

```python
from src.auth import service as auth_service
```

--------------------------------

### Resolve Constraint/Default Value Conflicts

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

A constraint like `ge=18` contradicts a `default=None`. Ensure constraints and default values are compatible.

```python
Field(ge=18, default=None)
```

--------------------------------

### Use Async Equivalents for Blocking Operations

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Avoid synchronous operations like `time.sleep`, `open()`, or sync database drivers within `async def` functions. Use their asynchronous counterparts such as `asyncio.sleep`, `aiofiles`, or an async database driver.

```python
asyncio.sleep
```

--------------------------------

### Document FastAPI endpoints with status codes and responses

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Fully document endpoints using decorators to specify response models, status codes, summaries, descriptions, tags, and detailed responses for various HTTP errors.

```python
from fastapi import APIRouter, status

router = APIRouter()


@router.post(
    "/items",
    response_model=ItemResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create an item",
    description="Creates an item owned by the authenticated user.",
    tags=["items"],
    responses={
        status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse, "description": "Validation error"},
        status.HTTP_409_CONFLICT:    {"model": ErrorResponse, "description": "Slug already exists"},
    },
)
async def create_item(payload: ItemCreate) -> ItemResponse: ...
```

--------------------------------

### FastAPI Quick Reference: Custom Datetime Serialization

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Implement custom datetime serialization using the `@field_serializer` decorator.

```python
@field_serializer
```

--------------------------------

### Hide Docs by Default in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Configure FastAPI to hide documentation by default, showing it only on specific environments like 'local' or 'staging'. This is achieved by conditionally setting `openapi_url` to `None` based on the environment variable.

```python
from fastapi import FastAPI
from starlette.config import Config

config = Config(".env")  # parse .env file for env variables

ENVIRONMENT = config("ENVIRONMENT")  # get current env name
SHOW_DOCS_ENVIRONMENT = ("local", "staging")  # explicit list of allowed envs

app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
   app_configs["openapi_url"] = None  # set url for docs as null

app = FastAPI(**app_configs)
```

--------------------------------

### Create a Custom Pydantic BaseModel for Serialization

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Implement a custom Pydantic BaseModel to enforce standard datetime formatting with timezone information and provide a method for serializing models into JSON-compatible dictionaries. This ensures consistent data output across the application.

```python
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, ConfigDict, field_serializer


class CustomModel(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    @field_serializer("*", when_used="json", check_fields=False)
    def _serialize_datetimes(self, value: Any) -> Any:
        if isinstance(value, datetime):
            if value.tzinfo is None:
                value = value.replace(tzinfo=ZoneInfo("UTC"))
            return value.strftime("%Y-%m-%dT%H:%M:%S%z")
        return value

    def serializable_dict(self, **kwargs):
        """Return a dict which contains only serializable fields."""
        default_dict = self.model_dump()

        return jsonable_encoder(default_dict)
```

--------------------------------

### Use Reliable Task Queues for Heavy Tasks

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Do not use `BackgroundTasks` for critical or potentially long-running operations, as they lack retries and die with the worker. Use robust task queue systems like Celery, Arq, or RQ.

```python
Celery / Arq / RQ
```

--------------------------------

### Specific Exception Handling in Routes

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Catching a broad `Exception` around a route's body can hide bugs and lead to silent 200s for 500 errors. Catch specific exception classes and raise `HTTPException` with a meaningful status.

```python
raise HTTPException with a meaningful status
```

--------------------------------

### Avoid Blocking I/O in Async Functions

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Do not use synchronous libraries like `requests` inside `async def` functions as it blocks the event loop. Use `httpx.AsyncClient` for async HTTP requests or `run_in_threadpool` for synchronous calls.

```python
await run_in_threadpool(requests.get, ...)
```

--------------------------------

### Use FastAPI BackgroundTasks for short, in-process tasks

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use BackgroundTasks for tasks that are expected to complete in under a second and can be silently dropped if the worker fails. These tasks run in the same worker process after the response is sent.

```python
from fastapi import BackgroundTasks

@router.post("/signup")
async def signup(data: SignupIn, bg: BackgroundTasks):
    user = await service.create_user(data)
    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process
    return user
```

--------------------------------

### Configuring OpenAPI Docs Visibility

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

This snippet shows how to conditionally disable OpenAPI documentation based on the environment. By default, docs are hidden unless the environment is explicitly listed in `SHOW_DOCS_ENVIRONMENT`.

```APIDOC
## Configuring OpenAPI Docs Visibility

### Description
This snippet shows how to conditionally disable OpenAPI documentation based on the environment. By default, docs are hidden unless the environment is explicitly listed in `SHOW_DOCS_ENVIRONMENT`.

### Code
```python
from fastapi import FastAPI
from starlette.config import Config

config = Config(".env")  # parse .env file for env variables

ENVIRONMENT = config("ENVIRONMENT")  # get current env name
SHOW_DOCS_ENVIRONMENT = ("local", "staging")  # explicit list of allowed envs

app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
   app_configs["openapi_url"] = None  # set url for docs as null

app = FastAPI(**app_configs)
```
```

--------------------------------

### Define and Use Reusable Dependencies in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates how to define and reuse dependencies like `valid_post_id`, `parse_jwt_data`, and `valid_owned_post`. FastAPI caches dependency results within a request, ensuring `parse_jwt_data` is called only once even if used multiple times.

```python
from fastapi.security import OAuth2PasswordBearer
import jwt  # PyJWT
from jwt.exceptions import InvalidTokenError

async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict[str, Any]:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except InvalidTokenError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: dict[str, Any] = Depends(valid_post_id), 
    token_data: dict[str, Any] = Depends(parse_jwt_data),
) -> dict[str, Any]:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post
```

```python
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)):
    return post
```

--------------------------------

### SQLAlchemy 2.0 Index Naming Convention

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Define a consistent naming convention for database indexes and constraints using SQLAlchemy's `MetaData`. This ensures clarity and uniformity across the database schema.

```python
from sqlalchemy import MetaData

POSTGRES_INDEXES_NAMING_CONVENTION = {
    "ix": "% (column_0_label)s_idx",
    "uq": "% (table_name)s_%(column_0_name)s_key",
    "ck": "% (table_name)s_%(constraint_name)s_check",
    "fk": "% (table_name)s_%(column_0_name)s_fkey",
    "pk": "% (table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
```

--------------------------------

### Conditionally disable OpenAPI docs in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Hide the automatic API documentation (Swagger UI and ReDoc) outside of specified environments by setting openapi_url to None during FastAPI app initialization.

```python
from fastapi import FastAPI
from src.config import settings

SHOW_DOCS_IN = {"local", "staging"}
app_kwargs = {"title": "My API"}
if settings.ENVIRONMENT not in SHOW_DOCS_IN:
    app_kwargs["openapi_url"] = None    # disables /docs and /redoc

app = FastAPI(**app_kwargs)
```

--------------------------------

### Use Async ORM Sessions

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Calling a synchronous ORM session within an `async def` function blocks the event loop and can deadlock the connection pool. Use `AsyncSession` for asynchronous database operations.

```python
AsyncSession
```

--------------------------------

### Use FastAPI Dependencies for Request Validation

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Leverage FastAPI's dependency injection system to perform complex request data validation, such as checking for the existence of a resource in the database before processing a request. This centralizes validation logic and avoids code duplication across endpoints.

```python
# dependencies.py
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


# router.py
@router.get("/posts/{post_id}", response_model=PostResponse)
async def get_post_by_id(post: dict[str, Any] = Depends(valid_post_id)):
    return post


@router.put("/posts/{post_id}", response_model=PostResponse)
async def update_post(
    update_data: PostUpdate,  
    post: dict[str, Any] = Depends(valid_post_id), 
):
    updated_post = await service.update(id=post["id"], data=update_data)
    return updated_post


@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)):
    post_reviews = await reviews_service.get_by_post_id(post["id"])
    return post_reviews
```

--------------------------------

### Importing Modules with Aliases in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use explicit module names for cross-domain imports to maintain clarity and avoid namespace conflicts. This pattern is recommended for organizing imports in FastAPI projects.

```python
from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode
```

--------------------------------

### Use Maintained JWT Library

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

The `python-jose` library is unmaintained. Use `PyJWT` for JWT operations.

```python
import jwt
```

--------------------------------

### FastAPI Quick Reference: Per-Request Dependency Caching

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Dependencies declared with `Depends` are cached per request by default, meaning the same dependency is only run once within a single request.

```python
Default behavior — same Depends(x) runs once
```

--------------------------------

### Modern FastAPI Dependency Injection with Annotated

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use `typing.Annotated` with `fastapi.Depends` for idiomatic dependency injection. This approach avoids potential issues with default argument values.

```python
# DO — modern Annotated form
from typing import Annotated
from fastapi import Depends

PostDep = Annotated[dict, Depends(valid_post_id)]

@router.get("/posts/{post_id}")
async def get_post(post: PostDep):
    return post
```

```python
# Avoid — default-argument form (still works, but legacy)
@router.get("/posts/{post_id}")
async def get_post(post: dict = Depends(valid_post_id)):
    return post
```

--------------------------------

### Pydantic Validation Error Handling in FastAPI

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Demonstrates how raising a `ValueError` within a Pydantic schema, when used in a request body, results in a `ValidationError` response from FastAPI. This helps provide clear feedback to users about invalid input.

```python
# src.profiles.schemas
from pydantic import BaseModel, field_validator

class ProfileCreate(BaseModel):
    username: str
    password: str
    
    @field_validator("password", mode="after")
    @classmethod
    def valid_password(cls, password: str) -> str:
        if not re.match(STRONG_PASSWORD_PATTERN, password):
            raise ValueError(
                "Password must contain at least "
                "one lower character, "
                "one upper character, "
                "digit or "
                "special symbol"
            )

        return password


# src.profiles.routes
from fastapi import APIRouter

router = APIRouter()


@router.post("/profiles")
async def create_profile(profile_data: ProfileCreate):
   pass
```

--------------------------------

### Avoid Double Model Construction

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Returning a Pydantic model and also setting `response_model=` to the same class causes the model to be constructed twice. Either return a dict/ORM row and let `response_model` validate, or omit `response_model` if the return type is trusted.

```python
return a dict/ORM row and let response_model validate
```

```python
drop response_model and trust the return type
```

--------------------------------

### Pydantic Model with Built-in Validators

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Defines a Pydantic model using built-in validators like `Field`, `EmailStr`, `AnyUrl`, and `StrEnum` for robust data validation. Pay attention to the interaction between constraints and default values.

```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field


class MusicBand(StrEnum):
    AEROSMITH = "AEROSMITH"
    QUEEN = "QUEEN"
    ACDC = "AC/DC"


class UserCreate(BaseModel):
    first_name: str = Field(min_length=1, max_length=128)
    username: str = Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_-]+$")
    email: EmailStr
    age: int = Field(ge=18)                     # required, must be >= 18
    favorite_band: MusicBand | None = None
    website: AnyUrl | None = None
```

--------------------------------

### Override FastAPI dependencies in tests

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Use FastAPI's dependency_overrides feature to replace dependencies with fakes during testing, avoiding the need for monkeypatching internal components.

```python
from src.auth.dependencies import parse_jwt_data
from src.main import app


def fake_user():
    return {"user_id": "00000000-0000-0000-0000-000000000001"}


@pytest.fixture(autouse=True)
def _override_auth():
    app.dependency_overrides[parse_jwt_data] = fake_user
    yield
    app.dependency_overrides.clear()
```

--------------------------------

### Define Pydantic Models for Post and Creator

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Defines Pydantic models for representing post data, including nested creator information. These models are used for data validation and serialization.

```python
from typing import Any

from pydantic import BaseModel, UUID4

   
class Creator(BaseModel):
    id: UUID4
    first_name: str
    last_name: str
    username: str


class Post(BaseModel):
    id: UUID4
    slug: str
    title: str
    creator: Creator
```

--------------------------------

### Pydantic v2 Field Serialization

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

The `model_config` with `json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` or `Annotated[T, PlainSerializer(...)]` for custom field serialization.

```python
@field_serializer
```

```python
Annotated[T, PlainSerializer(...)]
```

--------------------------------

### JWT Decoding with PyJWT

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/AGENTS.md

Decode JWT tokens securely using the `PyJWT` library. This function handles token validation and raises `InvalidTokenError` for malformed or invalid tokens.

```python
import jwt  # PyJWT
from jwt.exceptions import InvalidTokenError

def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])
    except InvalidTokenError as exc:
        raise InvalidCredentials() from exc
```

--------------------------------

### Define Pydantic Models with Custom Validation

Source: https://github.com/zhanymkanov/fastapi-best-practices/blob/master/README.md

Use Pydantic models to define data structures with built-in validation for fields like strings, emails, URLs, and enums. Enforces constraints such as minimum length, regex patterns, and age restrictions.

```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field


class MusicBand(StrEnum):
   AEROSMITH = "AEROSMITH"
   QUEEN = "QUEEN"
   ACDC = "AC/DC"


class UserBase(BaseModel):
    first_name: str = Field(min_length=1, max_length=128)
    username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$")
    email: EmailStr
    age: int = Field(ge=18)  # required, must be greater or equal to 18
    favorite_band: MusicBand | None = None  # only "AEROSMITH", "QUEEN", "AC/DC" values are allowed to be inputted
    website: AnyUrl | None = None
```
Tokens:
Raw