python-fastapi

What this skill does

## Quick Reference

| Layer | File | Purpose |
|-------|------|---------|
| Entrypoint | `main.py` | App factory, lifespan |
| Config | `config.py` | pydantic-settings |
| Routes | `api/v1/endpoints/` | Endpoint handlers |
| Schemas | `schemas/` | Pydantic models |
| Models | `models/` | SQLAlchemy models |
| Services | `services/` | Business logic |
| Deps | `api/deps.py` | Dependency injection |

| Pydantic Pattern | Use Case |
|------------------|----------|
| `UserCreate` | Input for creation |
| `UserUpdate` | Input for updates |
| `UserResponse` | API output |
| `UserInDB` | Internal with hash |

| Dependency | Code |
|------------|------|
| DB session | `db: Annotated[AsyncSession, Depends(get_db)]` |
| Current user | `user: Annotated[User, Depends(get_current_user)]` |
| Type alias | `CurrentUser = Annotated[User, Depends(...)]` |

| Production | Config |
|------------|--------|
| Server | `gunicorn -w 4 -k uvicorn.workers.UvicornWorker` |
| Workers | `CPU cores` for async |

## When to Use This Skill

Use for **FastAPI development**:
- Setting up FastAPI project structure
- Creating Pydantic schemas with validation
- Implementing dependency injection
- JWT authentication setup
- Docker deployment configuration

**Related skills:**
- For async patterns: see `python-asyncio`
- For testing: see `python-testing`
- For type hints: see `python-type-hints`

---

# FastAPI Production Best Practices (2025)

## Overview

FastAPI is a modern, high-performance web framework for building APIs. Built on Starlette and Pydantic, it provides automatic validation, serialization, and OpenAPI documentation.

## Project Structure

### Scalable Structure (Recommended)

```text
src/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application factory
│   ├── config.py            # Settings management
│   ├── database.py          # Database setup
│   ├── dependencies.py      # Shared dependencies
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py    # API router
│   │   │   └── endpoints/
│   │   │       ├── users.py
│   │   │       ├── items.py
│   │   │       └── auth.py
│   │   └── deps.py          # API dependencies
│   ├── core/
│   │   ├── __init__.py
│   │   ├── security.py      # Auth/JWT
│   │   └── exceptions.py    # Custom exceptions
│   ├── models/              # SQLAlchemy models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   └── services/            # Business logic
│       ├── __init__.py
│       ├── user.py
│       └── item.py
├── tests/
├── pyproject.toml
└── Dockerfile
```

## Configuration

### Settings with pydantic-settings

```python
# app/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # Application
    app_name: str = "My API"
    debug: bool = False
    environment: str = "production"

    # Database
    database_url: str
    database_pool_size: int = 5
    database_max_overflow: int = 10

    # Security
    secret_key: str
    access_token_expire_minutes: int = 30
    algorithm: str = "HS256"

    # External services
    redis_url: str | None = None


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

settings = get_settings()
```

### Application Factory

```python
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.api.v1.router import api_router
from app.database import engine, Base


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown
    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        openapi_url="/api/v1/openapi.json" if settings.debug else None,
        docs_url="/api/docs" if settings.debug else None,
        redoc_url="/api/redoc" if settings.debug else None,
        lifespan=lifespan,
    )

    # Middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"] if settings.debug else ["https://myapp.com"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Routes
    app.include_router(api_router, prefix="/api/v1")

    return app


app = create_app()
```

## Pydantic Schemas

### Input/Output Separation

```python
# app/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field

# Base schema with shared fields
class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = None

# Input schema for creation
class UserCreate(UserBase):
    password: str = Field(min_length=8)

# Input schema for updates
class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: str | None = None
    password: str | None = Field(None, min_length=8)

# Output schema (what API returns)
class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}

# Internal schema with password hash
class UserInDB(UserBase):
    id: int
    hashed_password: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}
```

### Validation Examples

```python
from typing import Annotated
from pydantic import BaseModel, Field, field_validator, model_validator

class ItemCreate(BaseModel):
    name: Annotated[str, Field(min_length=1, max_length=100)]
    price: Annotated[float, Field(gt=0, description="Price must be positive")]
    quantity: Annotated[int, Field(ge=0, le=10000)]
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be empty or whitespace")
        return v.strip()

    @field_validator("tags")
    @classmethod
    def tags_must_be_unique(cls, v: list[str]) -> list[str]:
        if len(v) != len(set(v)):
            raise ValueError("Tags must be unique")
        return v

    @model_validator(mode="after")
    def validate_price_quantity(self) -> "ItemCreate":
        if self.quantity > 0 and self.price < 0.01:
            raise ValueError("Price too low for available items")
        return self
```

## Async Best Practices

### Correct Async Usage

```python
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
import httpx

app = FastAPI()

# GOOD: Async database operations
async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# GOOD: Async HTTP calls
async def fetch_external_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# BAD: Blocking call in async route
@app.get("/bad")
async def bad_endpoint():
    import time
    time.sleep(5)  # Blocks entire event loop!
    return {"status": "done"}

# GOOD: Use asyncio.sleep or run in executor
@app.get("/good")
async def good_endpoint():
    import asyncio
    await asyncio.sleep(5)  # Non-blocking
    return {"status": "done"}

# GOOD: Run blocking code in executor
@app.get("/cpu-bound")
async def cpu_bound():
    import asyncio
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, heavy_computation)
    return {"result": result}
```

### Sync vs Async Routes

```python
# Async route - for I/O-bound operations
@app.get("/async-users")
async def get_users(db: AsyncSession = Depends(get_db