fastapi

Original source / Raw SKILL.md

---
name: fastapi
description: "FastAPI Python framework. Covers REST APIs, validation, dependencies, security. Keywords: Pydantic, async, OAuth2, JWT."
version: "0.128.0"
release_date: "2025-12-27"
---

# FastAPI

This skill provides comprehensive guidance for building APIs with FastAPI.

## Quick Navigation

| Topic              | Reference                           |
| ------------------ | ----------------------------------- |
| Getting started    | `references/first-steps.md`         |
| Path parameters    | `references/path-parameters.md`     |
| Query parameters   | `references/query-parameters.md`    |
| Request body       | `references/request-body.md`        |
| Validation         | `references/validation.md`          |
| Body advanced      | `references/body-advanced.md`       |
| Cookies/Headers    | `references/cookies-headers.md`     |
| Pydantic models    | `references/models.md`              |
| Forms/Files        | `references/forms-files.md`         |
| Error handling     | `references/error-handling.md`      |
| Path config        | `references/path-config.md`         |
| Dependencies       | `references/dependencies.md`        |
| Security           | `references/security.md`            |
| Middleware         | `references/middleware.md`          |
| CORS               | `references/cors.md`                |
| Database           | `references/sql-databases.md`       |
| Project structure  | `references/bigger-applications.md` |
| Background tasks   | `references/background-tasks.md`    |
| Metadata/Docs      | `references/metadata-docs.md`       |
| Testing            | `references/testing.md`             |
| Advanced responses | `references/responses-advanced.md`  |
| WebSockets         | `references/websockets.md`          |
| Templates          | `references/templates.md`           |
| Settings/Env vars  | `references/settings.md`            |
| Lifespan events    | `references/lifespan.md`            |
| OpenAPI advanced   | `references/openapi-advanced.md`    |

## When to Use

- Creating REST APIs with Python
- Adding endpoints with automatic validation
- Implementing OAuth2/JWT authentication
- Working with Pydantic models
- Adding dependency injection
- Configuring CORS, middleware
- Uploading files, handling forms
- Testing API endpoints

## Installation

```bash
pip install "fastapi[standard]"  # Full with uvicorn
pip install fastapi              # Minimal
pip install python-multipart     # For forms/files
```

## Quick Start

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}
```

Run: `fastapi dev main.py`

## Core Patterns

### Type-Safe Parameters

```python
from typing import Annotated
from fastapi import Path, Query

@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(ge=1)],
    q: Annotated[str | None, Query(max_length=50)] = None
):
    return {"item_id": item_id, "q": q}
```

### Request Body with Validation

```python
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)

@app.post("/items/", response_model=Item)
def create_item(item: Item):
    return item
```

### Dependencies

```python
from fastapi import Depends

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def list_users(db: Annotated[Session, Depends(get_db)]):
    return db.query(User).all()
```

### Authentication

```python
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    return decode_token(token)

@app.get("/users/me")
def read_me(user: Annotated[User, Depends(get_current_user)]):
    return user
```

## API Documentation

- Swagger UI: `/docs`
- ReDoc: `/redoc`
- OpenAPI: `/openapi.json`

## Best Practices

- Use `Annotated[Type, ...]` for parameters
- Define Pydantic models for request/response
- Use `response_model` for output filtering
- Add `status_code` for proper HTTP codes
- Use `tags` for API organization
- Add `dependencies` at router/app level for auth

## Prohibitions

- ❌ Return raw database models (use response models)
- ❌ Store passwords in plain text (use bcrypt/passlib)
- ❌ Mix `Body` with `Form`/`File` in same endpoint
- ❌ Use sync blocking I/O in async endpoints
- ❌ Skip HTTPException for error handling

## Links

- Docs: https://fastapi.tiangolo.com/
- Tutorial: https://fastapi.tiangolo.com/tutorial/
- Advanced: https://fastapi.tiangolo.com/advanced/


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### references/first-steps.md

```markdown
# First Steps

Core concepts for creating a minimal FastAPI application.

## Minimal Application

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}
```

## Key Components

### 1. FastAPI Instance

- `app = FastAPI()` creates the main application instance
- This is the entry point for all API functionality
- FastAPI inherits from Starlette (all Starlette features available)

### 2. Path Operation Decorator

- `@app.get("/")` defines a route handler
- Supports all HTTP methods:
  - `@app.get()` - read data
  - `@app.post()` - create data
  - `@app.put()` - update data
  - `@app.delete()` - delete data
  - `@app.patch()`, `@app.options()`, `@app.head()`, `@app.trace()`

### 3. Path Operation Function

- Can be `async def` or regular `def`
- Returns dict, list, str, int, Pydantic models
- Automatic JSON serialization

## Running the Server

```bash
fastapi dev main.py
```

Server runs at `http://127.0.0.1:8000`

## Auto-Generated Documentation

| URL             | Documentation            |
| --------------- | ------------------------ |
| `/docs`         | Swagger UI (interactive) |
| `/redoc`        | ReDoc (alternative)      |
| `/openapi.json` | Raw OpenAPI schema       |

## OpenAPI Integration

- FastAPI auto-generates OpenAPI 3.1.0 schema
- Schema includes paths, parameters, request/response models
- Powers interactive documentation
- Can generate client SDKs

## Terminology

| Term           | Meaning                                            |
| -------------- | -------------------------------------------------- |
| Path           | URL endpoint (e.g., `/items/foo`)                  |
| Operation      | HTTP method (GET, POST, etc.)                      |
| Path Operation | Combination of path + method                       |
| Schema         | Data structure definition (OpenAPI or JSON Schema) |

```

### references/path-parameters.md

```markdown
# Path Parameters

Capture dynamic values from URL paths.

## Basic Syntax

```python
@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}
```

## Type Annotations

```python
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
```

Benefits:

- **Data conversion**: String "3" → integer 3
- **Data validation**: Invalid values return clear error
- **Editor support**: Autocomplete, type checking
- **Auto documentation**: Types shown in OpenAPI docs

## Validation Error Response

```json
{
  "detail": [
    {
      "type": "int_parsing",
      "loc": ["path", "item_id"],
      "msg": "Input should be a valid integer",
      "input": "foo"
    }
  ]
}
```

## Order Matters

Fixed paths must be declared before parameterized paths:

```python
# ✅ Correct order
@app.get("/users/me")  # First - specific
async def read_user_me():
    return {"user_id": "current user"}

@app.get("/users/{user_id}")  # Second - generic
async def read_user(user_id: str):
    return {"user_id": user_id}
```

## Predefined Values with Enum

```python
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning"}
    return {"model_name": model_name}
```

Working with Enum values:

- Compare: `model_name is ModelName.alexnet`
- Get value: `model_name.value` → `"alexnet"`
- Returns JSON serialized string

## Path Parameters Containing Paths

Use `:path` converter for file paths:

```python
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}
```

URL: `/files//home/user/file.txt` (note double slash)

## Supported Types

- `str`, `int`, `float`, `bool`
- `UUID`
- Custom types via Pydantic

```

### references/query-parameters.md

```markdown
# Query Parameters

Parameters passed via URL query string (`?key=value&key2=value2`).

## Basic Usage

```python
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]
```

URL: `http://127.0.0.1:8000/items/?skip=20&limit=5`

## Parameter Types

### Required Parameters

```python
@app.get("/items/{item_id}")
async def read_item(item_id: str, needy: str):
    return {"item_id": item_id, "needy": needy}
```

No default = required. Missing parameter returns error.

### Optional Parameters

```python
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}
```

### Default Values

```python
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]
```

## Boolean Conversion

FastAPI auto-converts truthy strings to `bool`:

```python
@app.get("/items/{item_id}")
async def read_item(item_id: str, short: bool = False):
    ...
```

All these evaluate to `True`:

- `?short=1`
- `?short=True`
- `?short=true`
- `?short=on`
- `?short=yes`

## Mixed Parameters

Combine path and query parameters freely:

```python
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int,           # path
    item_id: str,           # path
    q: str | None = None,   # query, optional
    short: bool = False     # query, default
):
    ...
```

FastAPI distinguishes by:

- **Path parameters**: Declared in route path `{param}`
- **Query parameters**: Other function parameters

## Common Patterns

### Pagination

```python
@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 100):
    return db.query(Item).offset(skip).limit(limit).all()
```

### Filtering

```python
@app.get("/items/")
async def search_items(
    q: str | None = None,
    category: str | None = None,
    min_price: float | None = None
):
    ...
```

### Required + Optional Mix

```python
@app.get("/search/")
async def search(
    query: str,                    # required
    page: int = 1,                 # default
    per_page: int | None = None    # optional
):
    ...
```

```

### references/request-body.md

```markdown
# Request Body

Send JSON data from client to API using Pydantic models.

## Basic Example

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

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item
```

## Pydantic Model Rules

- Required fields: no default value
- Optional fields: `= None`
- Fields with defaults: `= value`

Valid JSON for above model:

```json
{"name": "Foo", "price": 45.2}
{"name": "Foo", "description": "A thing", "price": 45.2, "tax": 3.5}
```

## What FastAPI Does Automatically

1. Reads request body as JSON
2. Converts types (string → int, etc.)
3. Validates data structure
4. Returns clear errors for invalid data
5. Provides editor autocomplete for model attributes
6. Generates JSON Schema for OpenAPI docs

## Using the Model

```python
@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.model_dump()  # Convert to dict
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict
```

## Combining Parameters

FastAPI distinguishes parameter types automatically:

```python
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,              # Path parameter
    item: Item,                # Request body (Pydantic model)
    q: str | None = None       # Query parameter
):
    return {"item_id": item_id, **item.model_dump(), "q": q}
```

Recognition rules:

- Declared in path → **path parameter**
- Pydantic model → **request body**
- Singular type (str, int, etc.) → **query parameter**

## HTTP Methods for Body

- `POST` - Create (most common)
- `PUT` - Update/Replace
- `PATCH` - Partial update
- `DELETE` - Delete (rarely has body)

⚠️ `GET` with body is discouraged (undefined behavior in specs).

```

### references/validation.md

```markdown
# Parameter Validation

Advanced validation for Query, Path, Header, Cookie parameters.

## Query Validation with `Query`

```python
from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(max_length=50)] = None
):
    return {"q": q}
```

## String Validations

```python
q: Annotated[str | None, Query(
    min_length=3,
    max_length=50,
    pattern="^[a-z]+$"  # Regex pattern
)] = None
```

## Numeric Validations (Path/Query)

```python
from fastapi import Path

@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(ge=1, le=1000)],  # 1 ≤ value ≤ 1000
    size: Annotated[float, Query(gt=0, lt=100)]   # 0 < value < 100
):
    ...
```

| Parameter | Meaning               |
| --------- | --------------------- |
| `gt`      | Greater than          |
| `ge`      | Greater than or equal |
| `lt`      | Less than             |
| `le`      | Less than or equal    |

## Metadata for Documentation

```python
q: Annotated[str | None, Query(
    title="Search query",
    description="Full-text search in item names",
    min_length=3,
    max_length=50,
    deprecated=True  # Shows as deprecated in docs
)] = None
```

## Alias Parameter Name

```python
# URL: /items/?item-query=foo
@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(alias="item-query")] = None
):
    return {"q": q}
```

## Multiple Values (List)

```python
# URL: /items/?q=foo&q=bar
@app.get("/items/")
async def read_items(
    q: Annotated[list[str] | None, Query()] = None
):
    return {"q": q}  # ["foo", "bar"]
```

With defaults:

```python
q: Annotated[list[str], Query()] = ["default1", "default2"]
```

## Required Parameters with Validation

```python
# Required (no default)
q: Annotated[str, Query(min_length=3)]

# Required but can be None
q: Annotated[str | None, Query(min_length=3)]  # No default!
```

## Custom Validation

```python
from pydantic import AfterValidator

def validate_id(value: str) -> str:
    if not value.startswith(("isbn-", "imdb-")):
        raise ValueError("ID must start with isbn- or imdb-")
    return value

@app.get("/items/")
async def read_items(
    id: Annotated[str, AfterValidator(validate_id)]
):
    ...
```

## Exclude from OpenAPI

```python
hidden: Annotated[str | None, Query(include_in_schema=False)] = None
```

## Best Practice: Use Annotated

```python
# ✅ Recommended (Python 3.9+)
q: Annotated[str | None, Query(max_length=50)] = None

# ❌ Old style (avoid)
q: str | None = Query(default=None, max_length=50)
```

```

### references/body-advanced.md

```markdown
# Body - Advanced Topics

Advanced request body handling patterns.

## Multiple Body Parameters

Multiple Pydantic models = JSON keys by parameter names:

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

class Item(BaseModel):
    name: str
    price: float

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    return {"item_id": item_id, "item": item, "user": user}
```

Expected body:

```json
{
  "item": { "name": "Foo", "price": 42.0 },
  "user": { "username": "dave", "full_name": "Dave Grohl" }
}
```

## Singular Values in Body

Use `Body()` for non-model values in body:

```python
from typing import Annotated
from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: Annotated[int, Body()]
):
    return {"item_id": item_id, "importance": importance}
```

Expected body:

```json
{
    "item": {...},
    "user": {...},
    "importance": 5
}
```

## Embed Single Body Parameter

Force key wrapping for single model:

```python
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[Item, Body(embed=True)]
):
    return {"item_id": item_id, "item": item}
```

With `embed=True`:

```json
{ "item": { "name": "Foo", "price": 42.0 } }
```

Without `embed=True`:

```json
{ "name": "Foo", "price": 42.0 }
```

## Field Validation

Use `Field` from Pydantic for model attributes:

```python
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None,
        title="The description of the item",
        max_length=300
    )
    price: float = Field(gt=0, description="Must be greater than zero")
    tax: float | None = None
```

`Field` parameters: same as `Query`, `Path`, `Body`.

## Nested Models

Use Pydantic models as types in other models:

```python
from pydantic import BaseModel, HttpUrl

class Image(BaseModel):
    url: HttpUrl  # Validated URL
    name: str

class Item(BaseModel):
    name: str
    price: float
    images: list[Image] | None = None
```

## Special Types

- `set[str]` - Unique items, deduplicated
- `list[str]` - Regular list
- `dict[int, float]` - Keys convert from JSON strings
- `HttpUrl` - Validated URL
- `EmailStr` - Validated email (requires `pydantic[email]`)

## Bodies of Pure Lists

```python
@app.post("/images/multiple/")
async def create_multiple_images(images: list[Image]):
    return images
```

## Arbitrary Dict Bodies

```python
@app.post("/index-weights/")
async def create_index_weights(weights: dict[int, float]):
    return weights
```

JSON keys (strings) are converted to `int`.

## Extra Data Types

| Type                 | Request/Response Format |
| -------------------- | ----------------------- |
| `UUID`               | String                  |
| `datetime.datetime`  | ISO 8601 string         |
| `datetime.date`      | ISO 8601 string         |
| `datetime.time`      | ISO 8601 string         |
| `datetime.timedelta` | Float (total seconds)   |
| `bytes`              | String (binary format)  |
| `Decimal`            | Float                   |
| `frozenset`          | List (unique items)     |

```python
from datetime import datetime, timedelta
from uuid import UUID

@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: Annotated[datetime, Body()],
    process_after: Annotated[timedelta, Body()],
):
    start_process = start_datetime + process_after
    return {"item_id": item_id, "start_process": start_process}
```

```

### references/cookies-headers.md

```markdown
# Cookie and Header Parameters

Handling HTTP cookies and headers in FastAPI.

## Cookie Parameters

```python
from typing import Annotated
from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}
```

### Key Points

- **Import**: `from fastapi import Cookie`
- **Required**: Use `Cookie()` to distinguish from query params
- **Validation**: Same as `Query()`, `Path()`
- **Browser limitation**: Swagger UI cannot test cookies (JavaScript restriction)

### Optional Cookie

```python
@app.get("/items/")
async def read_items(
    session_id: Annotated[str | None, Cookie()] = None,
    tracking_id: Annotated[str | None, Cookie()] = None
):
    return {"session": session_id, "tracking": tracking_id}
```

## Header Parameters

```python
from typing import Annotated
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}
```

### Automatic Conversion

FastAPI converts:

- `user_agent` → `User-Agent`
- `x_token` → `X-Token`

```python
# Python variable: x_custom_header
# HTTP Header: X-Custom-Header (automatic conversion)
@app.get("/items/")
async def read_items(x_custom_header: Annotated[str | None, Header()] = None):
    return {"header": x_custom_header}
```

### Disable Underscore Conversion

```python
@app.get("/items/")
async def read_items(
    strange_header: Annotated[str | None, Header(convert_underscores=False)] = None
):
    return {"strange_header": strange_header}
```

⚠️ Warning: Some proxies/servers disallow headers with underscores.

### Duplicate Headers

For headers that can appear multiple times:

```python
@app.get("/items/")
async def read_items(x_token: Annotated[list[str] | None, Header()] = None):
    return {"X-Token values": x_token}
```

Request with:

```
X-Token: foo
X-Token: bar
```

Response:

```json
{ "X-Token values": ["bar", "foo"] }
```

## Common Headers

### Authorization

```python
@app.get("/protected/")
async def protected(authorization: Annotated[str | None, Header()] = None):
    if not authorization:
        raise HTTPException(401, "Missing authorization")
    # Parse "Bearer <token>"
    return {"auth": authorization}
```

### Content-Type

```python
@app.post("/webhook/")
async def webhook(
    content_type: Annotated[str | None, Header()] = None,
    body: bytes = Body(...)
):
    if content_type == "application/json":
        # Handle JSON
        pass
    return {"content_type": content_type}
```

### Custom Headers

```python
@app.get("/items/")
async def read_items(
    x_request_id: Annotated[str | None, Header()] = None,
    x_correlation_id: Annotated[str | None, Header()] = None
):
    return {"request_id": x_request_id, "correlation_id": x_correlation_id}
```

## Cookie/Header Parameter Models (FastAPI 0.115+)

### Cookie Model

```python
from pydantic import BaseModel

class Cookies(BaseModel):
    session_id: str
    tracking_id: str | None = None

@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
    return cookies
```

### Header Model

```python
class CommonHeaders(BaseModel):
    x_request_id: str
    x_correlation_id: str | None = None
    user_agent: str | None = None

@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
    return headers
```

## Setting Response Cookies/Headers

### Set Cookie

```python
from fastapi import Response

@app.post("/login/")
async def login(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=3600
    )
    return {"message": "logged in"}
```

### Set Header

```python
@app.get("/items/")
async def read_items(response: Response):
    response.headers["X-Custom-Header"] = "custom-value"
    return {"items": []}
```

## Recipes

### API Key in Header

```python
from fastapi import Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != "secret-api-key":
        raise HTTPException(403, "Invalid API key")
    return api_key

@app.get("/protected/", dependencies=[Depends(verify_api_key)])
async def protected_route():
    return {"message": "Access granted"}
```

### Request Tracing

```python
import uuid

@app.middleware("http")
async def add_request_id(request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response
```

## Comparison Table

| Feature          | Cookie       | Header              |
| ---------------- | ------------ | ------------------- |
| Import           | `Cookie`     | `Header`            |
| Auto-conversion  | No           | Underscore → Hyphen |
| Case-sensitive   | Yes          | No                  |
| Multiple values  | No           | Yes (list type)     |
| Browser testable | No (Swagger) | Yes                 |

```

### references/models.md

```markdown
````markdown
# Pydantic Models

Response models, multiple model patterns, and data transformation.

## Return Type Annotation

```python
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    tags: list[str] = []

@app.post("/items/")
async def create_item(item: Item) -> Item:
    return item

@app.get("/items/")
async def read_items() -> list[Item]:
    return [Item(name="Foo", price=42.0)]
```

FastAPI uses return type for:

- Data validation
- JSON Schema in OpenAPI
- Output data filtering

## response_model Parameter

Use when return type differs from actual response:

```python
class UserIn(BaseModel):
    username: str
    password: str  # Don't expose!
    email: str

class UserOut(BaseModel):
    username: str
    email: str

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user  # Password filtered out automatically
```

## Multiple Models Pattern

Different model states for different contexts:

```python
from pydantic import BaseModel, EmailStr

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

# Input model
class UserIn(UserBase):
    password: str

# Output model (no password)
class UserOut(UserBase):
    pass

# Database model
class UserInDB(UserBase):
    hashed_password: str

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    hashed = hash_password(user_in.password)
    user_db = UserInDB(**user_in.model_dump(), hashed_password=hashed)
    return user_db
```

## Model Conversion

```python
# Convert to dict
user_dict = user_in.model_dump()

# Create model with extra fields
UserInDB(**user_in.model_dump(), hashed_password=hashed)

# Partial copy with updates
updated = stored_model.model_copy(update=update_data)
```

## Response Filtering Options

```python
# Exclude unset values
@app.get("/items/{id}", response_model=Item, response_model_exclude_unset=True)

# Include only specific fields
@app.get("/items/{id}/name", response_model=Item, response_model_include={"name"})

# Exclude specific fields
@app.get("/items/{id}/public", response_model=Item, response_model_exclude={"tax"})

# Exclude defaults/None
response_model_exclude_defaults=True
response_model_exclude_none=True
```

## Union Types (Multiple Response Types)

```python
from typing import Union

class CarItem(BaseModel):
    type: str = "car"
    description: str

class PlaneItem(BaseModel):
    type: str = "plane"
    size: int
    description: str

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]
```

⚠️ Put more specific type first: `Union[PlaneItem, CarItem]`

## List and Dict Responses

```python
# List of models
@app.get("/items/", response_model=list[Item])
async def read_items():
    return items

# Arbitrary dict
@app.get("/weights/", response_model=dict[str, float])
async def read_weights():
    return {"foo": 2.3, "bar": 3.4}
```

## Return Response Directly

```python
from fastapi.responses import JSONResponse, RedirectResponse

@app.get("/portal")
async def get_portal(teleport: bool = False):
    if teleport:
        return RedirectResponse(url="https://example.com")
    return JSONResponse(content={"message": "Here"})

# Disable response model validation
@app.get("/portal", response_model=None)
async def get_portal():
    return {"message": "No validation"}
```

## CRUD Model Pattern

```python
class ItemBase(BaseModel):
    name: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None

class ItemInDB(ItemBase):
    id: int
    owner_id: int

class ItemResponse(ItemInDB):
    pass
```

## Partial Updates (PATCH)

```python
@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    stored = items[item_id]
    update_data = item.model_dump(exclude_unset=True)  # Only sent fields
    updated = stored.model_copy(update=update_data)
    items[item_id] = updated
    return updated
```

Key: `exclude_unset=True` includes only fields explicitly set by client.

## Prohibitions

- ❌ Don't store plaintext passwords in models
- ❌ Don't return input models with sensitive data
- ❌ Don't use `PlaneItem | CarItem` in response_model (use `Union[]`)
````

```

### references/forms-files.md

```markdown
# Forms and File Uploads

Form data handling and file uploads in FastAPI.

## Form Data

```python
from typing import Annotated
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    return {"username": username}
```

### Key Points

- **Dependency**: `pip install python-multipart`
- **Import**: `from fastapi import Form`
- **Encoding**: `application/x-www-form-urlencoded`
- **Validation**: Same as `Body()`, `Query()`, `Path()`

### OAuth2 Password Flow

```python
# OAuth2 spec requires exact field names
@app.post("/token")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    # Authenticate user
    return {"access_token": token}
```

## File Uploads

### bytes Parameter

```python
from typing import Annotated
from fastapi import FastAPI, File

@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}
```

- Entire file stored in memory
- Good for small files only

### UploadFile Parameter

```python
from fastapi import FastAPI, UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}
```

#### UploadFile Attributes

| Attribute      | Type                   | Description                    |
| -------------- | ---------------------- | ------------------------------ |
| `filename`     | `str`                  | Original filename              |
| `content_type` | `str`                  | MIME type (e.g., `image/jpeg`) |
| `file`         | `SpooledTemporaryFile` | Python file object             |

#### Async Methods

```python
contents = await myfile.read()      # Read all bytes
await myfile.seek(0)                # Go to start
await myfile.write(data)            # Write bytes
await myfile.close()                # Close file
```

#### Sync Access (in def functions)

```python
contents = myfile.file.read()
```

### Optional File Upload

```python
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No file sent"}
    return {"filename": file.filename}
```

### File with Metadata

```python
@app.post("/uploadfile/")
async def create_upload_file(
    file: Annotated[UploadFile, File(description="A file read as UploadFile")]
):
    return {"filename": file.filename}
```

## Multiple File Uploads

```python
from fastapi import FastAPI, File, UploadFile

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}

# With metadata
@app.post("/uploadfiles/")
async def create_upload_files(
    files: Annotated[list[UploadFile], File(description="Multiple files")]
):
    return {"filenames": [file.filename for file in files]}
```

## Forms and Files Together

```python
from fastapi import FastAPI, File, Form, UploadFile

@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()],
    fileb: Annotated[UploadFile, File()],
    token: Annotated[str, Form()]
):
    return {
        "file_size": len(file),
        "fileb_content_type": fileb.content_type,
        "token": token
    }
```

## Encoding Types

| Content Type                        | When Used           |
| ----------------------------------- | ------------------- |
| `application/x-www-form-urlencoded` | Forms without files |
| `multipart/form-data`               | Forms with files    |

## Prohibitions

- ❌ Cannot mix `Body` (JSON) with `Form`/`File` in same endpoint
- ❌ Don't use `bytes` for large files (memory)

## Recipes

### File Upload with Validation

```python
from fastapi import HTTPException

ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
MAX_SIZE = 5 * 1024 * 1024  # 5MB

@app.post("/upload/")
async def upload_file(file: UploadFile):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, "Invalid file type")

    contents = await file.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(400, "File too large")

    # Process file...
    return {"filename": file.filename}
```

### Save Uploaded File

```python
import shutil
from pathlib import Path

UPLOAD_DIR = Path("uploads")

@app.post("/upload/")
async def upload_file(file: UploadFile):
    file_path = UPLOAD_DIR / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename, "path": str(file_path)}
```

```

### references/error-handling.md

```markdown
# Error Handling

Return HTTP errors to clients with proper status codes.

## HTTPException

```python
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}
```

Key points:

- `raise` not `return` - terminates request immediately
- `detail` can be any JSON-serializable value (str, dict, list)

## Response Format

```json
{
  "detail": "Item not found"
}
```

## Custom Headers

```python
raise HTTPException(
    status_code=404,
    detail="Item not found",
    headers={"X-Error": "Custom header value"}
)
```

## Custom Exception Handler

```python
from fastapi import Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong"}
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}
```

## Override Validation Errors

```python
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
```

## Reuse Default Handlers

```python
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)

@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"HTTP error: {repr(exc)}")  # Log it
    return await http_exception_handler(request, exc)  # Use default
```

## Common Status Codes

| Code | Meaning        | Use Case           |
| ---- | -------------- | ------------------ |
| 400  | Bad Request    | Invalid input      |
| 401  | Unauthorized   | Missing auth       |
| 403  | Forbidden      | No permission      |
| 404  | Not Found      | Resource missing   |
| 409  | Conflict       | Duplicate resource |
| 422  | Unprocessable  | Validation failed  |
| 500  | Internal Error | Server bug         |

## FastAPI vs Starlette HTTPException

- FastAPI's `HTTPException` accepts any JSON-able `detail`
- Starlette's only accepts strings
- Register handlers for Starlette's version to catch all:

```python
from starlette.exceptions import HTTPException as StarletteHTTPException
```

```

### references/path-config.md

```markdown
# Path Operation Configuration

Metadata and configuration for endpoint documentation.

## Response Status Code

```python
from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

# Or use integer directly
@app.post("/items/", status_code=201)
async def create_item(item: Item):
    return item
```

Common codes: `200` OK, `201` Created, `204` No Content, `404` Not Found.

See `error-handling.md` for full status codes table.

## Tags

Group endpoints in docs:

```python
@app.post("/items/", tags=["items"])
async def create_item(item: Item):
    return item

@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]
```

### Tags with Enum

```python
from enum import Enum

class Tags(Enum):
    items = "items"
    users = "users"

@app.get("/items/", tags=[Tags.items])
async def get_items():
    return ["Portal gun", "Plumbus"]
```

## Summary and Description

```python
@app.post(
    "/items/",
    summary="Create an item",
    description="Create an item with all the information: name, description, price, tax, tags"
)
async def create_item(item: Item):
    return item
```

### Description from Docstring

```python
@app.post("/items/", summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item
```

Markdown supported in docstrings!

## Response Description

```python
@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    response_description="The created item"
)
async def create_item(item: Item):
    """Create an item with all the information."""
    return item
```

OpenAPI requires response description. FastAPI defaults to "Successful response" if omitted.

## Deprecate Endpoint

```python
@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]
```

Shown with strikethrough in docs.

## JSON Compatible Encoder

Convert Pydantic models to JSON-compatible dicts:

```python
from datetime import datetime
from fastapi.encoders import jsonable_encoder

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_data = jsonable_encoder(item)
    # datetime converted to ISO string
    fake_db[id] = json_compatible_data
```

### What jsonable_encoder Does

- `datetime` → ISO format string
- Pydantic model → dict
- Recursively converts nested objects
- Returns Python dict (not JSON string)

## Body Updates

### Full Update (PUT)

```python
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    stored = items[item_id]
    stored_model = Item(**stored)
    update_data = item.model_dump()
    updated = stored_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated)
    return updated
```

### Partial Update (PATCH)

```python
@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    stored = items[item_id]
    stored_model = Item(**stored)
    update_data = item.model_dump(exclude_unset=True)  # Only sent fields
    updated = stored_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated)
    return updated
```

Key: `exclude_unset=True` includes only fields explicitly set by client.

## All Configuration Parameters

```python
@app.post(
    "/items/",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    tags=["items"],
    summary="Create an item",
    description="Create a new item",
    response_description="The created item",
    deprecated=False,
    operation_id="create_item_items_post",
    include_in_schema=True,
    responses={
        201: {"description": "Created successfully"},
        400: {"description": "Bad request"}
    }
)
async def create_item(item: Item):
    return item
```

## OpenAPI Operation ID

```python
# Auto-generated: create_item_items_post
@app.post("/items/", operation_id="createItem")
async def create_item(item: Item):
    return item
```

Useful for SDK generation.

## Exclude from Schema

```python
@app.get("/internal/", include_in_schema=False)
async def internal_endpoint():
    return {"internal": "data"}
```

Not shown in OpenAPI docs.

```

### references/dependencies.md

```markdown
````markdown
# Dependencies

Dependency Injection system for shared logic, database connections, security.

## Basic Dependency

```python
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
```

## How It Works

1. Define a function (dependency) that takes parameters
2. Use `Depends(function)` in path operation parameters
3. FastAPI calls the dependency, gets result, passes to your function

## Reusable Type Alias

```python
CommonsDep = Annotated[dict, Depends(common_parameters)]

@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons
```

## Classes as Dependencies

```python
class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    # Shortcut: Depends() without argument when class == type hint
    return {"q": commons.q, "skip": commons.skip}
```

Any "callable" works: function, class, or object with `__call__`.

## Sub-dependencies

Dependencies can depend on other dependencies:

```python
def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None
):
    return q or last_query

@app.get("/items/")
async def read_query(query: Annotated[str, Depends(query_or_cookie_extractor)]):
    return {"q": query}
```

### use_cache Parameter

```python
# Default: use_cache=True - same instance reused in request
dep1: Annotated[dict, Depends(get_db)]
dep2: Annotated[dict, Depends(get_db)]  # Same instance as dep1

# Force new instance
dep2: Annotated[dict, Depends(get_db, use_cache=False)]
```

## Dependencies in Decorators

For side-effect dependencies (auth checks, logging) that don't return values:

```python
async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="Invalid token")

@app.get("/items/", dependencies=[Depends(verify_token)])
async def read_items():
    return [{"item": "Foo"}]
```

## Global Dependencies

Apply to all routes:

```python
app = FastAPI(dependencies=[Depends(verify_token)])
```

## Dependencies with yield

For setup/cleanup (e.g., database sessions):

```python
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items/")
async def read_items(db: Annotated[Session, Depends(get_db)]):
    return db.query(Item).all()
```

### Execution Order

1. Code before `yield` runs first
2. `yield db` passes value to endpoint
3. Endpoint executes
4. Code after `yield` runs (cleanup)

### Exception Handling

```python
async def get_db():
    db = DBSession()
    try:
        yield db
    except SomeException:
        db.rollback()
        raise
    finally:
        db.close()
```

⚠️ Cannot raise HTTPException AFTER yield - response already sent.

### Sub-dependencies with yield

Exit order is reverse of entry (LIFO):

```
Enter dependency_a → Enter dependency_b → Endpoint → Exit dependency_b → Exit dependency_a
```

## Recipes

### Database Session per Request

```python
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
def create_user(user: UserCreate, db: Annotated[Session, Depends(get_db)]):
    db_user = User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
```

### Authentication Chain

```python
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user = decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.get("/users/me")
async def read_users_me(user: Annotated[User, Depends(get_current_active_user)]):
    return user
```

### Parameterized Dependency

```python
def pagination_params(max_limit: int = 100):
    def inner(skip: int = 0, limit: int = 10):
        return {"skip": skip, "limit": min(limit, max_limit)}
    return inner

@app.get("/items/")
async def read_items(pagination: Annotated[dict, Depends(pagination_params(50))]):
    return pagination
```

### Callable Class Dependency

```python
class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False

checker = FixedContentQueryChecker(fixed_content="bar")

@app.get("/query-checker/")
async def check_query(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}
```

Instance with `__call__` allows storing state between requests.

## Use Cases

- Database sessions
- Authentication/Authorization
- Rate limiting
- Logging
- Shared query parameters
- Feature flags
- Configuration injection
````

```

### references/security.md

```markdown
# Security

Authentication and authorization patterns in FastAPI.

## Security Schemes (OpenAPI)

FastAPI supports standard OpenAPI security schemes:

| Scheme          | Description                       |
| --------------- | --------------------------------- |
| `apiKey`        | Key in query, header, or cookie   |
| `http`          | HTTP auth (Basic, Bearer, Digest) |
| `oauth2`        | OAuth2 flows                      |
| `openIdConnect` | OpenID Connect auto-discovery     |

## OAuth2 Password Flow (Simple)

```python
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    # Validate user credentials
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=400, detail="Incorrect credentials")
    return {"access_token": user.username, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: Annotated[str, Depends(oauth2_scheme)]):
    user = get_user_from_token(token)
    return user
```

## Get Current User Pattern

```python
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    user = decode_token(token)
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_user)]
):
    return current_user
```

## JWT Tokens

```python
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None
```

## Password Hashing

```python
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)
```

## HTTP Basic Auth

```python
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

security = HTTPBasic()

def get_current_username(
    credentials: Annotated[HTTPBasicCredentials, Depends(security)]
):
    # Use compare_digest to prevent timing attacks
    correct_user = secrets.compare_digest(
        credentials.username.encode("utf8"),
        b"admin"
    )
    correct_pass = secrets.compare_digest(
        credentials.password.encode("utf8"),
        b"secret"
    )
    if not (correct_user and correct_pass):
        raise HTTPException(
            status_code=401,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Basic"}
        )
    return credentials.username

@app.get("/admin")
def admin_area(username: Annotated[str, Depends(get_current_username)]):
    return {"message": f"Hello {username}"}
```

Important: Always use `secrets.compare_digest()` - regular `==` is vulnerable to timing attacks.

## OAuth2 Scopes

Fine-grained permissions with scopes:

```python
from fastapi import Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "items:read": "Read items",
        "items:write": "Create/update items",
        "users:read": "Read user info"
    }
)

async def get_current_user(
    security_scopes: SecurityScopes,
    token: Annotated[str, Depends(oauth2_scheme)]
):
    # Build WWW-Authenticate header
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"

    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value}
    )

    # Decode and verify token
    payload = verify_token(token)
    if payload is None:
        raise credentials_exception

    # Check scopes
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(
                status_code=403,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value}
            )

    return get_user(payload["sub"])

# Use Security() instead of Depends() for scopes
@app.get("/items/")
async def read_items(
    user: Annotated[User, Security(get_current_user, scopes=["items:read"])]
):
    return items

@app.post("/items/")
async def create_item(
    item: Item,
    user: Annotated[User, Security(get_current_user, scopes=["items:write"])]
):
    return item
```

## Token with Scopes

```python
def create_access_token(data: dict, scopes: list[str]):
    to_encode = data.copy()
    to_encode.update({
        "scopes": scopes,
        "exp": datetime.utcnow() + timedelta(minutes=30)
    })
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400)

    # Grant requested scopes (validate against user permissions)
    token = create_access_token(
        data={"sub": user.username},
        scopes=form_data.scopes
    )
    return {"access_token": token, "token_type": "bearer"}
```

```

### references/middleware.md

```markdown
# Middleware

Functions that process every request and response.

## How Middleware Works

1. Receive incoming request
2. Execute code before route handler
3. Pass request to route
4. Receive response from route
5. Execute code after route handler
6. Return response

## Create Middleware

```python
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response
```

## Middleware Function Parameters

- `request: Request` - The incoming request
- `call_next` - Function that passes request to route and returns response

## Execution Order

Middleware is stacked (last added = outermost):

```python
app.add_middleware(MiddlewareA)
app.add_middleware(MiddlewareB)
```

Execution:

- Request: B → A → route
- Response: route → A → B

## Common Patterns

### Request Logging

```python
@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response
```

### Authentication Check

```python
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    if request.url.path.startswith("/protected"):
        auth = request.headers.get("Authorization")
        if not auth:
            return JSONResponse(status_code=401, content={"detail": "Unauthorized"})
    return await call_next(request)
```

### Exception Handling

```python
@app.middleware("http")
async def catch_exceptions(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        return JSONResponse(status_code=500, content={"detail": str(e)})
```

## Built-in Middleware

### CORS

```python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
```

### GZip Compression

```python
from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)
```

### Trusted Host

```python
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)
```

### HTTPS Redirect

```python
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(HTTPSRedirectMiddleware)
```

## ASGI Middleware

Use any ASGI-compatible middleware:

```python
from unicorn import UnicornMiddleware

# Simple wrapping (not recommended)
# new_app = UnicornMiddleware(app, some_config="value")

# Recommended: use add_middleware for proper error handling
app.add_middleware(UnicornMiddleware, some_config="rainbow")
```

### Third-Party ASGI Middleware

- ProxyHeadersMiddleware (uvicorn)
- MessagePack middleware
- See: [ASGI Awesome List](https://github.com/florimondmanca/awesome-asgi)

## Timing Notes

- Dependencies with `yield`: exit code runs after middleware
- Background tasks: run after all middleware completes
- Use `time.perf_counter()` for precise timing (not `time.time()`)

## Custom Headers

- Prefix custom headers with `X-`
- For browser-visible headers, configure in CORS `expose_headers`

```

### references/cors.md

```markdown
# CORS (Cross-Origin Resource Sharing)

Enable cross-origin requests from frontend to backend.

## What is Origin?

Origin = protocol + domain + port

Different origins:

- `http://localhost` (port 80)
- `http://localhost:8080` (port 8080)
- `https://localhost` (HTTPS)

## Basic CORS Setup

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost",
    "http://localhost:3000",
    "http://localhost:8080",
    "https://myapp.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
```

## Configuration Options

| Parameter            | Default   | Description                                          |
| -------------------- | --------- | ---------------------------------------------------- |
| `allow_origins`      | `[]`      | List of allowed origins                              |
| `allow_origin_regex` | `None`    | Regex for origins (e.g., `https://.*\.example\.org`) |
| `allow_methods`      | `["GET"]` | Allowed HTTP methods                                 |
| `allow_headers`      | `[]`      | Allowed request headers                              |
| `allow_credentials`  | `False`   | Allow cookies/auth headers                           |
| `expose_headers`     | `[]`      | Headers visible to browser                           |
| `max_age`            | `600`     | Preflight cache time (seconds)                       |

## Common Configurations

### Development (Allow All)

```python
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
```

⚠️ Can't use `["*"]` with `allow_credentials=True`

### Production (Specific Origins)

```python
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://myapp.com",
        "https://www.myapp.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)
```

### Regex Pattern

```python
app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"https://.*\.myapp\.com",
)
```

## How CORS Works

### Preflight Requests

- Browser sends `OPTIONS` request first
- Includes `Origin` and `Access-Control-Request-Method` headers
- Backend responds with allowed origins/methods
- Browser then sends actual request

### Simple Requests

- GET, HEAD, POST with standard headers
- No preflight needed
- Backend adds CORS headers to response

## Credentials Warning

When `allow_credentials=True`:

- Cannot use `["*"]` for origins, methods, or headers
- Must explicitly list all allowed values

```python
# ❌ Invalid
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,  # Conflict!
)

# ✅ Valid
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
)
```

## Debugging CORS

Check browser console for:

- `Access-Control-Allow-Origin` header missing
- Origin not in allowed list
- Method not allowed
- Credential issues

Use browser DevTools Network tab to inspect preflight requests.

```

### references/sql-databases.md

```markdown
# SQL Databases

Use SQLModel with FastAPI for SQL database integration.

## Install

```bash
pip install sqlmodel
```

SQLModel = SQLAlchemy + Pydantic (same author as FastAPI).

## Basic Model

```python
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str
```

Key points:

- `table=True` = database table model
- `Field(primary_key=True)` = primary key
- `Field(index=True)` = create SQL index

## Engine Setup

```python
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}  # Required for SQLite
engine = create_engine(sqlite_url, connect_args=connect_args)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
```

## Session Dependency

```python
def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]
```

## Create Tables on Startup

```python
app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()
```

## CRUD Operations

### Create

```python
@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero
```

### Read List

```python
@app.get("/heroes/")
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes
```

### Read One

```python
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero
```

### Delete

```python
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}
```

## Multiple Models Pattern

Separate concerns with inheritance:

```python
# Base model (shared fields)
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

# Table model (database)
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

# Public model (API response - no secret_name)
class HeroPublic(HeroBase):
    id: int

# Create model (API input)
class HeroCreate(HeroBase):
    secret_name: str

# Update model (partial updates)
class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None
```

## Using Multiple Models

```python
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db
```

## Supported Databases

- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server
- Any SQLAlchemy-supported database

## Production Notes

- Use Alembic for migrations
- PostgreSQL recommended for production
- Check [Full Stack FastAPI Template](https://github.com/fastapi/full-stack-fastapi-template)

```

### references/bigger-applications.md

```markdown
# Bigger Applications - Multiple Files

Structure large FastAPI apps using APIRouter for modular organization.

## Project Structure

```
app/
├── __init__.py
├── main.py              # Main FastAPI app
├── dependencies.py      # Shared dependencies
├── routers/
│   ├── __init__.py
│   ├── users.py        # User routes
│   └── items.py        # Item routes
└── internal/
    ├── __init__.py
    └── admin.py        # Admin routes (shared)
```

## APIRouter

Create modular routers that work like mini FastAPI apps:

```python
# routers/users.py
from fastapi import APIRouter

router = APIRouter()

@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}
```

## Router with Prefix and Dependencies

Apply common settings to all routes in a router:

```python
# routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",               # All routes start with /items
    tags=["items"],               # OpenAPI tag
    dependencies=[Depends(get_token_header)],  # Auth for all routes
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def read_items():
    return fake_items_db

@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}
```

## Shared Dependencies

```python
# dependencies.py
from typing import Annotated
from fastapi import Header, HTTPException

async def get_token_header(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No token provided")
```

## Main App

Include routers in the main app:

```python
# main.py
from fastapi import Depends, FastAPI
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])  # Global dependency

# Include routers
app.include_router(users.router)
app.include_router(items.router)

# Include with custom prefix/dependencies
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}
```

## Relative Imports

```python
# From routers/items.py, import from parent package
from ..dependencies import get_token_header  # app/dependencies.py
```

- Single dot `.` = same package
- Double dots `..` = parent package
- Triple dots `...` = grandparent (rarely needed)

## Include Router in Router

```python
# Nest routers before including in main app
router.include_router(other_router)
```

## Multiple Prefixes for Same Router

```python
# Same router at different paths (e.g., /api/v1 and /api/latest)
app.include_router(api_router, prefix="/api/v1")
app.include_router(api_router, prefix="/api/latest")
```

## Key Points

- `APIRouter` = mini `FastAPI` class with same parameters
- `prefix` must not end with `/`
- Router dependencies execute before decorator dependencies
- Path operations are "cloned" (not mounted) to include in OpenAPI schema
- Performance: including routers happens at startup (microseconds)

```

### references/background-tasks.md

```markdown
# Background Tasks

Run tasks after returning the response.

## Use Cases

- Send email notifications
- Process uploaded files
- Update caches
- Log analytics

## Basic Usage

```python
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message: str = ""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}
```

## Add Task Parameters

```python
background_tasks.add_task(
    function,       # Task function
    arg1,           # Positional args
    arg2,
    key1=value1,    # Keyword args
    key2=value2,
)
```

## Task Function Types

- Use `def` for I/O-bound tasks (file write, sync calls)
- Use `async def` for async operations (FastAPI handles both)

```python
# Sync task
def process_file(path: str):
    with open(path) as f:
        # process...

# Async task
async def call_external_api(url: str):
    async with httpx.AsyncClient() as client:
        await client.get(url)
```

## With Dependency Injection

```python
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q

@app.post("/send-notification/{email}")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
    q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}
```

All tasks from dependencies and path operations are merged and run after response.

## Heavy Background Jobs

For complex workloads, consider:

- **Celery** - Distributed task queue with RabbitMQ/Redis
- **Redis Queue (RQ)** - Simple Redis-based queue
- **Dramatiq** - Alternative to Celery

Use these when you need:

- Tasks running on multiple processes/servers
- Retries and failure handling
- Scheduled/periodic tasks
- Task result tracking

## When to Use BackgroundTasks

✅ Good for:

- Small, quick tasks
- Access to same app variables/objects
- No need for separate worker process

❌ Better use Celery for:

- Long-running computations
- Tasks needing multiple workers/servers
- Complex retry logic

## Import Note

```python
# Use BackgroundTasks (plural) not BackgroundTask
from fastapi import BackgroundTasks  # ✓
# NOT from starlette.background import BackgroundTask
```

```

### references/metadata-docs.md

```markdown
# API Metadata and Documentation URLs

Customize OpenAPI metadata and documentation UI.

## API Metadata

```python
from fastapi import FastAPI

description = """
ChimichangApp API helps you do awesome stuff. 🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""

app = FastAPI(
    title="ChimichangApp",
    description=description,
    summary="Deadpool's favorite app. Nuff said.",
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "[email protected]",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)
```

### Metadata Fields

| Field              | Type | Description                           |
| ------------------ | ---- | ------------------------------------- |
| `title`            | str  | API title                             |
| `summary`          | str  | Short summary (OpenAPI 3.1.0+)        |
| `description`      | str  | Full description (Markdown supported) |
| `version`          | str  | Your API version (e.g., "2.5.0")      |
| `terms_of_service` | str  | URL to ToS                            |
| `contact`          | dict | Contact info                          |
| `license_info`     | dict | License info                          |

### License Identifier (OpenAPI 3.1.0+)

```python
license_info={
    "name": "Apache 2.0",
    "identifier": "MIT",  # Instead of URL
}
```

## Tag Metadata

```python
tags_metadata = [
    {
        "name": "users",
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]

@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]
```

Tag order in docs follows list order (not alphabetical).

## OpenAPI URL

```python
# Default: /openapi.json
app = FastAPI(openapi_url="/api/v1/openapi.json")

# Disable OpenAPI entirely
app = FastAPI(openapi_url=None)
```

## Docs URLs

```python
# Defaults: /docs (Swagger), /redoc (ReDoc)
app = FastAPI(
    docs_url="/documentation",  # Swagger UI
    redoc_url="/redocumentation",  # ReDoc
)

# Disable docs
app = FastAPI(docs_url=None, redoc_url=None)
```

## Static Files

```python
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")
```

### Mount Parameters

| Parameter            | Description               |
| -------------------- | ------------------------- |
| `"/static"`          | URL path prefix           |
| `directory="static"` | Local directory           |
| `name="static"`      | Internal name for FastAPI |

### What is Mounting?

Mounting adds a complete independent application at a specific path. Mounted apps are:

- Completely independent
- Not included in main app's OpenAPI
- Handle all sub-paths themselves

## Debugging

### VS Code / PyCharm Debug Setup

```python
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

### VS Code Debug Config

1. Go to "Debug" panel
2. "Add configuration..." → Python
3. Select "Python: Current File (Integrated Terminal)"
4. Set breakpoints and run

### PyCharm Debug

1. Open "Run" menu
2. Select "Debug..."
3. Choose the file (e.g., `main.py`)

### About `__name__ == "__main__"`

- Runs code only when file is executed directly: `python myapp.py`
- Does NOT run when imported: `from myapp import app`

## Recipes

### Production vs Development Docs

```python
import os

ENVIRONMENT = os.getenv("ENVIRONMENT", "development")

app = FastAPI(
    docs_url="/docs" if ENVIRONMENT == "development" else None,
    redoc_url="/redoc" if ENVIRONMENT == "development" else None,
)
```

### Custom OpenAPI Schema

```python
from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title="Custom Title",
        version="1.0.0",
        description="Custom description",
        routes=app.routes,
    )
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi
```

### Serve Static with HTML

```python
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def read_index():
    return FileResponse("static/index.html")
```

```

### references/testing.md

```markdown
# Testing

Test FastAPI applications using TestClient and pytest.

## Setup

```bash
pip install httpx pytest
```

## Basic Test

```python
from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}
```

## Key Points

- Test functions use `def`, not `async def`
- Client calls are synchronous (no `await`)
- TestClient is from Starlette, re-exported by FastAPI

## Project Structure

```
app/
├── __init__.py
├── main.py
└── test_main.py
```

```python
# test_main.py
from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
```

## Request Methods

```python
# GET with query params
response = client.get("/items/?skip=0&limit=10")

# POST with JSON body
response = client.post(
    "/items/",
    json={"name": "Foo", "price": 42.0}
)

# With headers
response = client.get(
    "/items/foo",
    headers={"X-Token": "secret-token"}
)

# With cookies
response = client.get(
    "/items/",
    cookies={"session": "abc123"}
)

# Form data
response = client.post(
    "/login/",
    data={"username": "user", "password": "pass"}
)

# File upload
with open("file.txt", "rb") as f:
    response = client.post(
        "/upload/",
        files={"file": ("filename.txt", f, "text/plain")}
    )
```

## Testing Errors

```python
def test_read_item_not_found():
    response = client.get("/items/nonexistent")
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}

def test_create_item_invalid():
    response = client.post(
        "/items/",
        json={"name": "Foo"}  # Missing required field
    )
    assert response.status_code == 422  # Validation error
```

## Override Dependencies

```python
from fastapi.testclient import TestClient
from app.main import app, get_db

def override_get_db():
    return TestDatabase()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_with_mock_db():
    response = client.get("/items/")
    assert response.status_code == 200

# Reset overrides after test
app.dependency_overrides = {}
```

## Override Settings

```python
from .config import Settings
from .main import app, get_settings

def get_settings_override():
    return Settings(admin_email="[email protected]")

app.dependency_overrides[get_settings] = get_settings_override
```

## Async Tests

For async database calls or when testing async code:

```python
import pytest
from httpx import ASGITransport, AsyncClient
from .main import app

@pytest.mark.anyio
async def test_async():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
```

Requires: `pip install anyio pytest-anyio httpx`

Note: `AsyncClient` doesn't trigger lifespan events. Use `asgi-lifespan` if needed:

```python
from asgi_lifespan import LifespanManager

@pytest.mark.anyio
async def test_with_lifespan():
    async with LifespanManager(app):
        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://test"
        ) as ac:
            response = await ac.get("/")
```

## Testing WebSockets

```python
def test_websocket():
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("hello")
        data = websocket.receive_text()
        assert data == "Echo: hello"
```

## Testing Lifespan Events

```python
def test_app_lifespan():
    with TestClient(app) as client:
        # Startup runs when entering context
        response = client.get("/")
        assert response.status_code == 200
    # Shutdown runs when exiting context
```

## Run Tests

```bash
pytest
pytest -v  # Verbose
pytest test_main.py::test_read_main  # Specific test
pytest -x  # Stop on first failure
pytest --tb=short  # Shorter traceback
```

```