{"owner":"zhanymkanov","repo":"fastapi-best-practices","hasSkills":true,"totalSkillsCount":1,"totalTokensCount":4108,"categories":["root-instruction"],"hasMcp":false,"mcpConfig":null,"found":["AGENTS.md"],"skills":{"AGENTS.md":"# FastAPI Best Practices for AI Agents\n\nA machine-readable companion to [README.md](./README.md) for AI coding agents\nworking in FastAPI projects. Same rules, restructured for fast pattern matching:\nversion pins, Do/Don't blocks, anti-patterns, and a quick-reference table.\n\n## Compatibility Matrix\n\nPin to these versions or newer. Examples in this file assume them.\n\n| Dependency       | Minimum   | Notes                                                |\n|------------------|-----------|------------------------------------------------------|\n| Python           | 3.11      | Required for `StrEnum` and `X \\| Y` union syntax     |\n| FastAPI          | 0.115     | `Annotated[T, Depends(...)]` is the idiomatic form   |\n| Pydantic         | 2.7       | v1 APIs (`json_encoders`, `.dict()`) are removed     |\n| pydantic-settings| 2.4       | Lives in a separate package since Pydantic v2        |\n| SQLAlchemy       | 2.0       | Use the async API (`AsyncSession`, `async_sessionmaker`) |\n| Alembic          | 1.13      | Async-aware migrations                               |\n| httpx            | 0.27      | Use `ASGITransport` for in-process tests             |\n| PyJWT            | 2.9       | Use this, not the unmaintained `python-jose`         |\n| ruff             | 0.6       | Replaces black, isort, autoflake                     |\n\n## Project Structure\n\nOrganize by domain, not by file type. One package per bounded context.\n\n```\nsrc/\n├── {domain}/           # e.g., auth/, posts/, aws/\n│   ├── router.py       # API endpoints\n│   ├── schemas.py      # Pydantic models\n│   ├── models.py       # SQLAlchemy ORM models\n│   ├── service.py      # Business logic\n│   ├── dependencies.py # Route dependencies\n│   ├── config.py       # Domain-scoped BaseSettings\n│   ├── constants.py    # Constants and error codes\n│   ├── exceptions.py   # Domain-specific exceptions\n│   └── utils.py        # Helper functions\n├── config.py           # Global BaseSettings\n├── models.py           # Shared Pydantic / ORM bases\n├── exceptions.py       # Global exceptions\n├── database.py         # Async engine + session factory\n└── main.py             # FastAPI app + lifespan\n```\n\n**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`.\n\n```python\nfrom src.auth import constants as auth_constants\nfrom src.notifications import service as notification_service\nfrom src.posts.constants import ErrorCode as PostsErrorCode\n```\n\n## Async Routes\n\n### Decision rule\n\n| Route does this                        | Use         |\n|----------------------------------------|-------------|\n| `await`-able non-blocking I/O          | `async def` |\n| Blocking I/O (no async client exists)  | `def` (sync, runs in threadpool) |\n| Mix of both                            | `async def` + `run_in_threadpool` for the blocking part |\n| CPU-bound work (>50 ms compute)        | Offload to a worker process (Celery / RQ / Arq) |\n\n### Do / Don't\n\n```python\n# DON'T — blocking call inside async route freezes the entire event loop\n@router.get(\"/bad\")\nasync def bad():\n    time.sleep(10)            # blocks every request on this worker\n    return {\"ok\": True}\n\n# DO — sync route lets FastAPI run it in a threadpool\n@router.get(\"/sync-ok\")\ndef sync_ok():\n    time.sleep(10)            # blocks one threadpool worker, not the loop\n    return {\"ok\": True}\n\n# DO — async route with awaitable sleep\n@router.get(\"/async-ok\")\nasync def async_ok():\n    await asyncio.sleep(10)   # yields control, loop keeps serving requests\n    return {\"ok\": True}\n\n# DO — async route that has to call a sync library\nfrom fastapi.concurrency import run_in_threadpool\n\n@router.get(\"/wrap\")\nasync def wrap():\n    result = await run_in_threadpool(legacy_sync_client.fetch, \"id\")\n    return result\n```\n\n### Threadpool caveats\n- Default Starlette threadpool size is 40. Saturating it slows every sync route.\n- Threads cost more than coroutines. Don't use sync routes \"just because.\"\n\n## Pydantic\n\n### Use built-in validators\n```python\nfrom enum import StrEnum\nfrom pydantic import AnyUrl, BaseModel, EmailStr, Field\n\n\nclass MusicBand(StrEnum):\n    AEROSMITH = \"AEROSMITH\"\n    QUEEN = \"QUEEN\"\n    ACDC = \"AC/DC\"\n\n\nclass UserCreate(BaseModel):\n    first_name: str = Field(min_length=1, max_length=128)\n    username: str = Field(min_length=1, max_length=128, pattern=r\"^[A-Za-z0-9_-]+$\")\n    email: EmailStr\n    age: int = Field(ge=18)                     # required, must be >= 18\n    favorite_band: MusicBand | None = None\n    website: AnyUrl | None = None\n```\n\n> **Don't** write `Field(ge=18, default=None)`. The constraint and the default contradict\n> each other. Decide: required (`Field(ge=18)`) or optional (`int | None = Field(default=None, ge=18)`).\n\n### Custom base model — modern serialization\n\n`json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` for per-field rules,\nor annotate a custom type with `PlainSerializer`.\n\n```python\nfrom datetime import datetime\nfrom zoneinfo import ZoneInfo\nfrom pydantic import BaseModel, ConfigDict, field_serializer\n\n\nclass CustomModel(BaseModel):\n    model_config = ConfigDict(populate_by_name=True)\n\n    @field_serializer(\"*\", when_used=\"json\", check_fields=False)\n    def _serialize_datetimes(self, value):\n        if isinstance(value, datetime):\n            if value.tzinfo is None:\n                value = value.replace(tzinfo=ZoneInfo(\"UTC\"))\n            return value.strftime(\"%Y-%m-%dT%H:%M:%S%z\")\n        return value\n```\n\n### Split BaseSettings by domain\n\n`pydantic-settings` is its own package since Pydantic v2.\n\n```python\n# src/auth/config.py\nfrom datetime import timedelta\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\n\n\nclass AuthConfig(BaseSettings):\n    model_config = SettingsConfigDict(env_prefix=\"AUTH_\", env_file=\".env\", extra=\"ignore\")\n\n    JWT_ALG: str\n    JWT_SECRET: str\n    JWT_EXP_MINUTES: int = 5\n    REFRESH_TOKEN_KEY: str\n    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)\n    SECURE_COOKIES: bool = True\n\n\nauth_settings = AuthConfig()\n```\n\n## Dependencies\n\n### Use Annotated, not default-arg `Depends(...)`\n\n`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids\ngotchas with default values.\n\n```python\n# DO — modern Annotated form\nfrom typing import Annotated\nfrom fastapi import Depends\n\nPostDep = Annotated[dict, Depends(valid_post_id)]\n\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: PostDep):\n    return post\n\n# Avoid — default-argument form (still works, but legacy)\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: dict = Depends(valid_post_id)):\n    return post\n```\n\n### Validate inside dependencies (not just inject)\n```python\nasync def valid_post_id(post_id: UUID4) -> dict:\n    post = await service.get_by_id(post_id)\n    if not post:\n        raise PostNotFound()\n    return post\n```\n\n### Chain dependencies for reuse\n```python\nasync def valid_owned_post(\n    post: Annotated[dict, Depends(valid_post_id)],\n    token_data: Annotated[dict, Depends(parse_jwt_data)],\n) -> dict:\n    if post[\"creator_id\"] != token_data[\"user_id\"]:\n        raise UserNotOwner()\n    return post\n```\n\n### Rules\n- Dependencies are **cached per request**. Same `Depends(x)` called 5 times in one request → `x` runs once.\n- Prefer `async def` dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks.\n- Use **the same path-variable name** across endpoints when you want to share a dependency (e.g. `profile_id` in both `/profiles/{profile_id}` and `/creators/{profile_id}`).\n\n## Authentication — JWT\n\nUse **`PyJWT`**, not `python-jose` (unmaintained).\n\n```python\nimport jwt  # PyJWT\nfrom jwt.exceptions import InvalidTokenError\n\ndef decode_token(token: str) -> dict:\n    try:\n        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])\n    except InvalidTokenError as exc:\n        raise InvalidCredentials() from exc\n```\n\n## Database — SQLAlchemy 2.0 async\n\nPrefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.\n\n```python\n# src/database.py\nfrom sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine\n\nengine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)\nSessionFactory = async_sessionmaker(engine, expire_on_commit=False)\n\n\nasync def get_db() -> AsyncSession:\n    async with SessionFactory() as session:\n        yield session\n```\n\n### Naming conventions\n- `lower_case_snake`\n- Singular tables: `post`, `user`, `post_like`\n- Group with prefix: `payment_account`, `payment_bill`\n- `_at` suffix for `datetime`, `_date` suffix for `date`\n- Use the same FK column name everywhere it appears (`profile_id`, not `user_id` in some tables and `profile_id` in others)\n\n### Index naming convention\n```python\nfrom sqlalchemy import MetaData\n\nPOSTGRES_INDEXES_NAMING_CONVENTION = {\n    \"ix\": \"%(column_0_label)s_idx\",\n    \"uq\": \"%(table_name)s_%(column_0_name)s_key\",\n    \"ck\": \"%(table_name)s_%(constraint_name)s_check\",\n    \"fk\": \"%(table_name)s_%(column_0_name)s_fkey\",\n    \"pk\": \"%(table_name)s_pkey\",\n}\nmetadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)\n```\n\n### SQL-first, Pydantic-second\n- Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this.\n- Hydrate the result into Pydantic only for response validation, not for transformation.\n\n## Background work — BackgroundTasks vs Celery\n\n| Use BackgroundTasks when…                | Use Celery / Arq / RQ when…                |\n|------------------------------------------|--------------------------------------------|\n| Task is < 1 second                       | Task takes seconds to minutes              |\n| Failure can be silently dropped          | You need retries, dead-letter, or visibility|\n| Task is in-process (send email, log row) | Task is CPU-heavy or needs a separate pool |\n| You don't need scheduling                | You need cron, ETA, or rate limiting       |\n\n```python\nfrom fastapi import BackgroundTasks\n\n@router.post(\"/signup\")\nasync def signup(data: SignupIn, bg: BackgroundTasks):\n    user = await service.create_user(data)\n    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process\n    return user\n```\n\n> BackgroundTasks run **after the response is sent, in the same worker process**. If the\n> worker dies, the task is lost. There is no retry. Don't use them for anything you'd\n> page on.\n\n## Testing\n\n### Async client from day one\n```python\nimport pytest\nfrom httpx import AsyncClient, ASGITransport\n\nfrom src.main import app\n\n\n@pytest.fixture\nasync def client():\n    transport = ASGITransport(app=app)\n    async with AsyncClient(transport=transport, base_url=\"http://test\") as ac:\n        yield ac\n\n\n@pytest.mark.asyncio\nasync def test_create_post(client: AsyncClient):\n    resp = await client.post(\"/posts\", json={\"title\": \"hi\"})\n    assert resp.status_code == 201\n```\n\n> **Don't** use `async_asgi_testclient` — it's unmaintained. The example above (httpx +\n> `ASGITransport`) is the supported path.\n\n### Override dependencies in tests\nDon't monkeypatch internals. Use FastAPI's built-in `dependency_overrides`.\n\n```python\nfrom src.auth.dependencies import parse_jwt_data\nfrom src.main import app\n\n\ndef fake_user():\n    return {\"user_id\": \"00000000-0000-0000-0000-000000000001\"}\n\n\n@pytest.fixture(autouse=True)\ndef _override_auth():\n    app.dependency_overrides[parse_jwt_data] = fake_user\n    yield\n    app.dependency_overrides.clear()\n```\n\n## Migrations (Alembic)\n\n- Migrations must be static and reversible.\n- Use the async template: `alembic init -t async migrations`\n- Descriptive filenames:\n  ```ini\n  # alembic.ini\n  file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s\n  ```\n  → `2026-04-14_add_post_content_idx.py`\n\n## API documentation\n\n### Hide docs outside selected envs\n```python\nfrom fastapi import FastAPI\nfrom src.config import settings\n\nSHOW_DOCS_IN = {\"local\", \"staging\"}\napp_kwargs = {\"title\": \"My API\"}\nif settings.ENVIRONMENT not in SHOW_DOCS_IN:\n    app_kwargs[\"openapi_url\"] = None    # disables /docs and /redoc\n\napp = FastAPI(**app_kwargs)\n```\n\n### Document endpoints fully\n```python\nfrom fastapi import APIRouter, status\n\nrouter = APIRouter()\n\n\n@router.post(\n    \"/items\",\n    response_model=ItemResponse,\n    status_code=status.HTTP_201_CREATED,\n    summary=\"Create an item\",\n    description=\"Creates an item owned by the authenticated user.\",\n    tags=[\"items\"],\n    responses={\n        status.HTTP_400_BAD_REQUEST: {\"model\": ErrorResponse, \"description\": \"Validation error\"},\n        status.HTTP_409_CONFLICT:    {\"model\": ErrorResponse, \"description\": \"Slug already exists\"},\n    },\n)\nasync def create_item(payload: ItemCreate) -> ItemResponse: ...\n```\n\n## Linting\n\n```shell\nruff check --fix src\nruff format src\n```\n\nAdd to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.\n\n---\n\n## Anti-patterns — common AI-agent mistakes\n\nIf you're an agent reviewing a diff, check for these. Each is a real failure mode I've\nseen agents introduce.\n\n| Anti-pattern | Why it's wrong | Fix |\n|---|---|---|\n| `requests.get(...)` inside `async def` | Blocks the event loop. `requests` is sync. | Use `httpx.AsyncClient` or `await run_in_threadpool(requests.get, ...)`. |\n| `time.sleep` / `open()` / sync DB driver inside `async def` | Same — blocks the loop. | Use the async equivalent (`asyncio.sleep`, `aiofiles`, async driver). |\n| `from jose import jwt` | `python-jose` is unmaintained. | `import jwt` (PyJWT). |\n| `from async_asgi_testclient import TestClient` | Unmaintained. | `httpx.AsyncClient` + `ASGITransport`. |\n| `model_config = ConfigDict(json_encoders={...})` | Deprecated in Pydantic v2. | `@field_serializer` or `Annotated[T, PlainSerializer(...)]`. |\n| `Field(ge=18, default=None)` | Constraint contradicts the default. | Pick required or optional, not both. |\n| `def get_user(id: int = Depends(...))` (default-arg form) | Legacy; gotchas with default values. | `user: Annotated[User, Depends(...)]`. |\n| Catching `Exception` around a route's body | Hides bugs and turns 500s into silent 200s. | Catch the specific exception class; raise `HTTPException` with a meaningful status. |\n| `BackgroundTasks` for anything you'd page on | No retry, dies with the worker. | Use Celery / Arq / RQ. |\n| Calling a sync ORM session inside `async def` | Blocks the loop, may deadlock the pool. | Use `AsyncSession`. |\n| Returning a Pydantic model and *also* setting `response_model=` to that same class | Model gets constructed twice (validate + serialize). | Either return a `dict`/ORM row and let `response_model` validate, or drop `response_model` and trust the return type. |\n| Importing across domains via deep paths (`from src.auth.service.user import ...`) | Tight coupling, hard to refactor. | `from src.auth import service as auth_service`. |\n| Reusing one `BaseSettings` for the whole app | Hard to reason about, every domain reads every var. | One `BaseSettings` per domain. |\n| Mocking the database in integration tests | Mock/prod divergence eventually fires in prod. | Use a real DB (testcontainers, ephemeral schema) and `dependency_overrides` for auth/external services. |\n\n## Quick reference\n\n| Scenario                             | Solution                                          |\n|--------------------------------------|---------------------------------------------------|\n| Non-blocking I/O                     | `async def` route with `await`                    |\n| Blocking I/O (no async client)       | `def` route (sync, runs in threadpool)            |\n| Sync library inside async route      | `await run_in_threadpool(fn, *args)`              |\n| CPU-intensive work                   | Celery / Arq / RQ worker process                  |\n| Request validation against DB        | Dependency that loads + validates + returns       |\n| Reuse validation across routes       | Chain dependencies                                |\n| Inject dependency in modern style    | `Annotated[T, Depends(...)]`                      |\n| Per-request dep caching              | Default behavior — same `Depends(x)` runs once    |\n| Per-domain config                    | One `BaseSettings` subclass per domain            |\n| Custom datetime serialization        | `@field_serializer`                               |\n| Fire-and-forget short task           | `BackgroundTasks`                                 |\n| Reliable / scheduled / heavy task    | Celery / Arq / RQ                                 |\n| JWT decode                           | `PyJWT` (`import jwt`)                            |\n| Async DB                             | SQLAlchemy 2.0 async (`AsyncSession`)             |\n| HTTP test client                     | `httpx.AsyncClient` + `ASGITransport`             |\n| Swap dep in tests                    | `app.dependency_overrides[dep] = fake`            |\n| Lint + format                        | `ruff check --fix` + `ruff format`                |\n"},"files":{"AGENTS.md":"# FastAPI Best Practices for AI Agents\n\nA machine-readable companion to [README.md](./README.md) for AI coding agents\nworking in FastAPI projects. Same rules, restructured for fast pattern matching:\nversion pins, Do/Don't blocks, anti-patterns, and a quick-reference table.\n\n## Compatibility Matrix\n\nPin to these versions or newer. Examples in this file assume them.\n\n| Dependency       | Minimum   | Notes                                                |\n|------------------|-----------|------------------------------------------------------|\n| Python           | 3.11      | Required for `StrEnum` and `X \\| Y` union syntax     |\n| FastAPI          | 0.115     | `Annotated[T, Depends(...)]` is the idiomatic form   |\n| Pydantic         | 2.7       | v1 APIs (`json_encoders`, `.dict()`) are removed     |\n| pydantic-settings| 2.4       | Lives in a separate package since Pydantic v2        |\n| SQLAlchemy       | 2.0       | Use the async API (`AsyncSession`, `async_sessionmaker`) |\n| Alembic          | 1.13      | Async-aware migrations                               |\n| httpx            | 0.27      | Use `ASGITransport` for in-process tests             |\n| PyJWT            | 2.9       | Use this, not the unmaintained `python-jose`         |\n| ruff             | 0.6       | Replaces black, isort, autoflake                     |\n\n## Project Structure\n\nOrganize by domain, not by file type. One package per bounded context.\n\n```\nsrc/\n├── {domain}/           # e.g., auth/, posts/, aws/\n│   ├── router.py       # API endpoints\n│   ├── schemas.py      # Pydantic models\n│   ├── models.py       # SQLAlchemy ORM models\n│   ├── service.py      # Business logic\n│   ├── dependencies.py # Route dependencies\n│   ├── config.py       # Domain-scoped BaseSettings\n│   ├── constants.py    # Constants and error codes\n│   ├── exceptions.py   # Domain-specific exceptions\n│   └── utils.py        # Helper functions\n├── config.py           # Global BaseSettings\n├── models.py           # Shared Pydantic / ORM bases\n├── exceptions.py       # Global exceptions\n├── database.py         # Async engine + session factory\n└── main.py             # FastAPI app + lifespan\n```\n\n**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`.\n\n```python\nfrom src.auth import constants as auth_constants\nfrom src.notifications import service as notification_service\nfrom src.posts.constants import ErrorCode as PostsErrorCode\n```\n\n## Async Routes\n\n### Decision rule\n\n| Route does this                        | Use         |\n|----------------------------------------|-------------|\n| `await`-able non-blocking I/O          | `async def` |\n| Blocking I/O (no async client exists)  | `def` (sync, runs in threadpool) |\n| Mix of both                            | `async def` + `run_in_threadpool` for the blocking part |\n| CPU-bound work (>50 ms compute)        | Offload to a worker process (Celery / RQ / Arq) |\n\n### Do / Don't\n\n```python\n# DON'T — blocking call inside async route freezes the entire event loop\n@router.get(\"/bad\")\nasync def bad():\n    time.sleep(10)            # blocks every request on this worker\n    return {\"ok\": True}\n\n# DO — sync route lets FastAPI run it in a threadpool\n@router.get(\"/sync-ok\")\ndef sync_ok():\n    time.sleep(10)            # blocks one threadpool worker, not the loop\n    return {\"ok\": True}\n\n# DO — async route with awaitable sleep\n@router.get(\"/async-ok\")\nasync def async_ok():\n    await asyncio.sleep(10)   # yields control, loop keeps serving requests\n    return {\"ok\": True}\n\n# DO — async route that has to call a sync library\nfrom fastapi.concurrency import run_in_threadpool\n\n@router.get(\"/wrap\")\nasync def wrap():\n    result = await run_in_threadpool(legacy_sync_client.fetch, \"id\")\n    return result\n```\n\n### Threadpool caveats\n- Default Starlette threadpool size is 40. Saturating it slows every sync route.\n- Threads cost more than coroutines. Don't use sync routes \"just because.\"\n\n## Pydantic\n\n### Use built-in validators\n```python\nfrom enum import StrEnum\nfrom pydantic import AnyUrl, BaseModel, EmailStr, Field\n\n\nclass MusicBand(StrEnum):\n    AEROSMITH = \"AEROSMITH\"\n    QUEEN = \"QUEEN\"\n    ACDC = \"AC/DC\"\n\n\nclass UserCreate(BaseModel):\n    first_name: str = Field(min_length=1, max_length=128)\n    username: str = Field(min_length=1, max_length=128, pattern=r\"^[A-Za-z0-9_-]+$\")\n    email: EmailStr\n    age: int = Field(ge=18)                     # required, must be >= 18\n    favorite_band: MusicBand | None = None\n    website: AnyUrl | None = None\n```\n\n> **Don't** write `Field(ge=18, default=None)`. The constraint and the default contradict\n> each other. Decide: required (`Field(ge=18)`) or optional (`int | None = Field(default=None, ge=18)`).\n\n### Custom base model — modern serialization\n\n`json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` for per-field rules,\nor annotate a custom type with `PlainSerializer`.\n\n```python\nfrom datetime import datetime\nfrom zoneinfo import ZoneInfo\nfrom pydantic import BaseModel, ConfigDict, field_serializer\n\n\nclass CustomModel(BaseModel):\n    model_config = ConfigDict(populate_by_name=True)\n\n    @field_serializer(\"*\", when_used=\"json\", check_fields=False)\n    def _serialize_datetimes(self, value):\n        if isinstance(value, datetime):\n            if value.tzinfo is None:\n                value = value.replace(tzinfo=ZoneInfo(\"UTC\"))\n            return value.strftime(\"%Y-%m-%dT%H:%M:%S%z\")\n        return value\n```\n\n### Split BaseSettings by domain\n\n`pydantic-settings` is its own package since Pydantic v2.\n\n```python\n# src/auth/config.py\nfrom datetime import timedelta\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\n\n\nclass AuthConfig(BaseSettings):\n    model_config = SettingsConfigDict(env_prefix=\"AUTH_\", env_file=\".env\", extra=\"ignore\")\n\n    JWT_ALG: str\n    JWT_SECRET: str\n    JWT_EXP_MINUTES: int = 5\n    REFRESH_TOKEN_KEY: str\n    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)\n    SECURE_COOKIES: bool = True\n\n\nauth_settings = AuthConfig()\n```\n\n## Dependencies\n\n### Use Annotated, not default-arg `Depends(...)`\n\n`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids\ngotchas with default values.\n\n```python\n# DO — modern Annotated form\nfrom typing import Annotated\nfrom fastapi import Depends\n\nPostDep = Annotated[dict, Depends(valid_post_id)]\n\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: PostDep):\n    return post\n\n# Avoid — default-argument form (still works, but legacy)\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: dict = Depends(valid_post_id)):\n    return post\n```\n\n### Validate inside dependencies (not just inject)\n```python\nasync def valid_post_id(post_id: UUID4) -> dict:\n    post = await service.get_by_id(post_id)\n    if not post:\n        raise PostNotFound()\n    return post\n```\n\n### Chain dependencies for reuse\n```python\nasync def valid_owned_post(\n    post: Annotated[dict, Depends(valid_post_id)],\n    token_data: Annotated[dict, Depends(parse_jwt_data)],\n) -> dict:\n    if post[\"creator_id\"] != token_data[\"user_id\"]:\n        raise UserNotOwner()\n    return post\n```\n\n### Rules\n- Dependencies are **cached per request**. Same `Depends(x)` called 5 times in one request → `x` runs once.\n- Prefer `async def` dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks.\n- Use **the same path-variable name** across endpoints when you want to share a dependency (e.g. `profile_id` in both `/profiles/{profile_id}` and `/creators/{profile_id}`).\n\n## Authentication — JWT\n\nUse **`PyJWT`**, not `python-jose` (unmaintained).\n\n```python\nimport jwt  # PyJWT\nfrom jwt.exceptions import InvalidTokenError\n\ndef decode_token(token: str) -> dict:\n    try:\n        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])\n    except InvalidTokenError as exc:\n        raise InvalidCredentials() from exc\n```\n\n## Database — SQLAlchemy 2.0 async\n\nPrefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.\n\n```python\n# src/database.py\nfrom sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine\n\nengine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)\nSessionFactory = async_sessionmaker(engine, expire_on_commit=False)\n\n\nasync def get_db() -> AsyncSession:\n    async with SessionFactory() as session:\n        yield session\n```\n\n### Naming conventions\n- `lower_case_snake`\n- Singular tables: `post`, `user`, `post_like`\n- Group with prefix: `payment_account`, `payment_bill`\n- `_at` suffix for `datetime`, `_date` suffix for `date`\n- Use the same FK column name everywhere it appears (`profile_id`, not `user_id` in some tables and `profile_id` in others)\n\n### Index naming convention\n```python\nfrom sqlalchemy import MetaData\n\nPOSTGRES_INDEXES_NAMING_CONVENTION = {\n    \"ix\": \"%(column_0_label)s_idx\",\n    \"uq\": \"%(table_name)s_%(column_0_name)s_key\",\n    \"ck\": \"%(table_name)s_%(constraint_name)s_check\",\n    \"fk\": \"%(table_name)s_%(column_0_name)s_fkey\",\n    \"pk\": \"%(table_name)s_pkey\",\n}\nmetadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)\n```\n\n### SQL-first, Pydantic-second\n- Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this.\n- Hydrate the result into Pydantic only for response validation, not for transformation.\n\n## Background work — BackgroundTasks vs Celery\n\n| Use BackgroundTasks when…                | Use Celery / Arq / RQ when…                |\n|------------------------------------------|--------------------------------------------|\n| Task is < 1 second                       | Task takes seconds to minutes              |\n| Failure can be silently dropped          | You need retries, dead-letter, or visibility|\n| Task is in-process (send email, log row) | Task is CPU-heavy or needs a separate pool |\n| You don't need scheduling                | You need cron, ETA, or rate limiting       |\n\n```python\nfrom fastapi import BackgroundTasks\n\n@router.post(\"/signup\")\nasync def signup(data: SignupIn, bg: BackgroundTasks):\n    user = await service.create_user(data)\n    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process\n    return user\n```\n\n> BackgroundTasks run **after the response is sent, in the same worker process**. If the\n> worker dies, the task is lost. There is no retry. Don't use them for anything you'd\n> page on.\n\n## Testing\n\n### Async client from day one\n```python\nimport pytest\nfrom httpx import AsyncClient, ASGITransport\n\nfrom src.main import app\n\n\n@pytest.fixture\nasync def client():\n    transport = ASGITransport(app=app)\n    async with AsyncClient(transport=transport, base_url=\"http://test\") as ac:\n        yield ac\n\n\n@pytest.mark.asyncio\nasync def test_create_post(client: AsyncClient):\n    resp = await client.post(\"/posts\", json={\"title\": \"hi\"})\n    assert resp.status_code == 201\n```\n\n> **Don't** use `async_asgi_testclient` — it's unmaintained. The example above (httpx +\n> `ASGITransport`) is the supported path.\n\n### Override dependencies in tests\nDon't monkeypatch internals. Use FastAPI's built-in `dependency_overrides`.\n\n```python\nfrom src.auth.dependencies import parse_jwt_data\nfrom src.main import app\n\n\ndef fake_user():\n    return {\"user_id\": \"00000000-0000-0000-0000-000000000001\"}\n\n\n@pytest.fixture(autouse=True)\ndef _override_auth():\n    app.dependency_overrides[parse_jwt_data] = fake_user\n    yield\n    app.dependency_overrides.clear()\n```\n\n## Migrations (Alembic)\n\n- Migrations must be static and reversible.\n- Use the async template: `alembic init -t async migrations`\n- Descriptive filenames:\n  ```ini\n  # alembic.ini\n  file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s\n  ```\n  → `2026-04-14_add_post_content_idx.py`\n\n## API documentation\n\n### Hide docs outside selected envs\n```python\nfrom fastapi import FastAPI\nfrom src.config import settings\n\nSHOW_DOCS_IN = {\"local\", \"staging\"}\napp_kwargs = {\"title\": \"My API\"}\nif settings.ENVIRONMENT not in SHOW_DOCS_IN:\n    app_kwargs[\"openapi_url\"] = None    # disables /docs and /redoc\n\napp = FastAPI(**app_kwargs)\n```\n\n### Document endpoints fully\n```python\nfrom fastapi import APIRouter, status\n\nrouter = APIRouter()\n\n\n@router.post(\n    \"/items\",\n    response_model=ItemResponse,\n    status_code=status.HTTP_201_CREATED,\n    summary=\"Create an item\",\n    description=\"Creates an item owned by the authenticated user.\",\n    tags=[\"items\"],\n    responses={\n        status.HTTP_400_BAD_REQUEST: {\"model\": ErrorResponse, \"description\": \"Validation error\"},\n        status.HTTP_409_CONFLICT:    {\"model\": ErrorResponse, \"description\": \"Slug already exists\"},\n    },\n)\nasync def create_item(payload: ItemCreate) -> ItemResponse: ...\n```\n\n## Linting\n\n```shell\nruff check --fix src\nruff format src\n```\n\nAdd to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.\n\n---\n\n## Anti-patterns — common AI-agent mistakes\n\nIf you're an agent reviewing a diff, check for these. Each is a real failure mode I've\nseen agents introduce.\n\n| Anti-pattern | Why it's wrong | Fix |\n|---|---|---|\n| `requests.get(...)` inside `async def` | Blocks the event loop. `requests` is sync. | Use `httpx.AsyncClient` or `await run_in_threadpool(requests.get, ...)`. |\n| `time.sleep` / `open()` / sync DB driver inside `async def` | Same — blocks the loop. | Use the async equivalent (`asyncio.sleep`, `aiofiles`, async driver). |\n| `from jose import jwt` | `python-jose` is unmaintained. | `import jwt` (PyJWT). |\n| `from async_asgi_testclient import TestClient` | Unmaintained. | `httpx.AsyncClient` + `ASGITransport`. |\n| `model_config = ConfigDict(json_encoders={...})` | Deprecated in Pydantic v2. | `@field_serializer` or `Annotated[T, PlainSerializer(...)]`. |\n| `Field(ge=18, default=None)` | Constraint contradicts the default. | Pick required or optional, not both. |\n| `def get_user(id: int = Depends(...))` (default-arg form) | Legacy; gotchas with default values. | `user: Annotated[User, Depends(...)]`. |\n| Catching `Exception` around a route's body | Hides bugs and turns 500s into silent 200s. | Catch the specific exception class; raise `HTTPException` with a meaningful status. |\n| `BackgroundTasks` for anything you'd page on | No retry, dies with the worker. | Use Celery / Arq / RQ. |\n| Calling a sync ORM session inside `async def` | Blocks the loop, may deadlock the pool. | Use `AsyncSession`. |\n| Returning a Pydantic model and *also* setting `response_model=` to that same class | Model gets constructed twice (validate + serialize). | Either return a `dict`/ORM row and let `response_model` validate, or drop `response_model` and trust the return type. |\n| Importing across domains via deep paths (`from src.auth.service.user import ...`) | Tight coupling, hard to refactor. | `from src.auth import service as auth_service`. |\n| Reusing one `BaseSettings` for the whole app | Hard to reason about, every domain reads every var. | One `BaseSettings` per domain. |\n| Mocking the database in integration tests | Mock/prod divergence eventually fires in prod. | Use a real DB (testcontainers, ephemeral schema) and `dependency_overrides` for auth/external services. |\n\n## Quick reference\n\n| Scenario                             | Solution                                          |\n|--------------------------------------|---------------------------------------------------|\n| Non-blocking I/O                     | `async def` route with `await`                    |\n| Blocking I/O (no async client)       | `def` route (sync, runs in threadpool)            |\n| Sync library inside async route      | `await run_in_threadpool(fn, *args)`              |\n| CPU-intensive work                   | Celery / Arq / RQ worker process                  |\n| Request validation against DB        | Dependency that loads + validates + returns       |\n| Reuse validation across routes       | Chain dependencies                                |\n| Inject dependency in modern style    | `Annotated[T, Depends(...)]`                      |\n| Per-request dep caching              | Default behavior — same `Depends(x)` runs once    |\n| Per-domain config                    | One `BaseSettings` subclass per domain            |\n| Custom datetime serialization        | `@field_serializer`                               |\n| Fire-and-forget short task           | `BackgroundTasks`                                 |\n| Reliable / scheduled / heavy task    | Celery / Arq / RQ                                 |\n| JWT decode                           | `PyJWT` (`import jwt`)                            |\n| Async DB                             | SQLAlchemy 2.0 async (`AsyncSession`)             |\n| HTTP test client                     | `httpx.AsyncClient` + `ASGITransport`             |\n| Swap dep in tests                    | `app.dependency_overrides[dep] = fake`            |\n| Lint + format                        | `ruff check --fix` + `ruff format`                |\n"},"items":[{"name":"AGENTS.md","path":"AGENTS.md","rawUrl":"https://raw.githubusercontent.com/zhanymkanov/fastapi-best-practices/HEAD/AGENTS.md","title":"AI Agent Protocol & Instructions","category":"root-instruction","format":"markdown","content":"# FastAPI Best Practices for AI Agents\n\nA machine-readable companion to [README.md](./README.md) for AI coding agents\nworking in FastAPI projects. Same rules, restructured for fast pattern matching:\nversion pins, Do/Don't blocks, anti-patterns, and a quick-reference table.\n\n## Compatibility Matrix\n\nPin to these versions or newer. Examples in this file assume them.\n\n| Dependency       | Minimum   | Notes                                                |\n|------------------|-----------|------------------------------------------------------|\n| Python           | 3.11      | Required for `StrEnum` and `X \\| Y` union syntax     |\n| FastAPI          | 0.115     | `Annotated[T, Depends(...)]` is the idiomatic form   |\n| Pydantic         | 2.7       | v1 APIs (`json_encoders`, `.dict()`) are removed     |\n| pydantic-settings| 2.4       | Lives in a separate package since Pydantic v2        |\n| SQLAlchemy       | 2.0       | Use the async API (`AsyncSession`, `async_sessionmaker`) |\n| Alembic          | 1.13      | Async-aware migrations                               |\n| httpx            | 0.27      | Use `ASGITransport` for in-process tests             |\n| PyJWT            | 2.9       | Use this, not the unmaintained `python-jose`         |\n| ruff             | 0.6       | Replaces black, isort, autoflake                     |\n\n## Project Structure\n\nOrganize by domain, not by file type. One package per bounded context.\n\n```\nsrc/\n├── {domain}/           # e.g., auth/, posts/, aws/\n│   ├── router.py       # API endpoints\n│   ├── schemas.py      # Pydantic models\n│   ├── models.py       # SQLAlchemy ORM models\n│   ├── service.py      # Business logic\n│   ├── dependencies.py # Route dependencies\n│   ├── config.py       # Domain-scoped BaseSettings\n│   ├── constants.py    # Constants and error codes\n│   ├── exceptions.py   # Domain-specific exceptions\n│   └── utils.py        # Helper functions\n├── config.py           # Global BaseSettings\n├── models.py           # Shared Pydantic / ORM bases\n├── exceptions.py       # Global exceptions\n├── database.py         # Async engine + session factory\n└── main.py             # FastAPI app + lifespan\n```\n\n**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`.\n\n```python\nfrom src.auth import constants as auth_constants\nfrom src.notifications import service as notification_service\nfrom src.posts.constants import ErrorCode as PostsErrorCode\n```\n\n## Async Routes\n\n### Decision rule\n\n| Route does this                        | Use         |\n|----------------------------------------|-------------|\n| `await`-able non-blocking I/O          | `async def` |\n| Blocking I/O (no async client exists)  | `def` (sync, runs in threadpool) |\n| Mix of both                            | `async def` + `run_in_threadpool` for the blocking part |\n| CPU-bound work (>50 ms compute)        | Offload to a worker process (Celery / RQ / Arq) |\n\n### Do / Don't\n\n```python\n# DON'T — blocking call inside async route freezes the entire event loop\n@router.get(\"/bad\")\nasync def bad():\n    time.sleep(10)            # blocks every request on this worker\n    return {\"ok\": True}\n\n# DO — sync route lets FastAPI run it in a threadpool\n@router.get(\"/sync-ok\")\ndef sync_ok():\n    time.sleep(10)            # blocks one threadpool worker, not the loop\n    return {\"ok\": True}\n\n# DO — async route with awaitable sleep\n@router.get(\"/async-ok\")\nasync def async_ok():\n    await asyncio.sleep(10)   # yields control, loop keeps serving requests\n    return {\"ok\": True}\n\n# DO — async route that has to call a sync library\nfrom fastapi.concurrency import run_in_threadpool\n\n@router.get(\"/wrap\")\nasync def wrap():\n    result = await run_in_threadpool(legacy_sync_client.fetch, \"id\")\n    return result\n```\n\n### Threadpool caveats\n- Default Starlette threadpool size is 40. Saturating it slows every sync route.\n- Threads cost more than coroutines. Don't use sync routes \"just because.\"\n\n## Pydantic\n\n### Use built-in validators\n```python\nfrom enum import StrEnum\nfrom pydantic import AnyUrl, BaseModel, EmailStr, Field\n\n\nclass MusicBand(StrEnum):\n    AEROSMITH = \"AEROSMITH\"\n    QUEEN = \"QUEEN\"\n    ACDC = \"AC/DC\"\n\n\nclass UserCreate(BaseModel):\n    first_name: str = Field(min_length=1, max_length=128)\n    username: str = Field(min_length=1, max_length=128, pattern=r\"^[A-Za-z0-9_-]+$\")\n    email: EmailStr\n    age: int = Field(ge=18)                     # required, must be >= 18\n    favorite_band: MusicBand | None = None\n    website: AnyUrl | None = None\n```\n\n> **Don't** write `Field(ge=18, default=None)`. The constraint and the default contradict\n> each other. Decide: required (`Field(ge=18)`) or optional (`int | None = Field(default=None, ge=18)`).\n\n### Custom base model — modern serialization\n\n`json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` for per-field rules,\nor annotate a custom type with `PlainSerializer`.\n\n```python\nfrom datetime import datetime\nfrom zoneinfo import ZoneInfo\nfrom pydantic import BaseModel, ConfigDict, field_serializer\n\n\nclass CustomModel(BaseModel):\n    model_config = ConfigDict(populate_by_name=True)\n\n    @field_serializer(\"*\", when_used=\"json\", check_fields=False)\n    def _serialize_datetimes(self, value):\n        if isinstance(value, datetime):\n            if value.tzinfo is None:\n                value = value.replace(tzinfo=ZoneInfo(\"UTC\"))\n            return value.strftime(\"%Y-%m-%dT%H:%M:%S%z\")\n        return value\n```\n\n### Split BaseSettings by domain\n\n`pydantic-settings` is its own package since Pydantic v2.\n\n```python\n# src/auth/config.py\nfrom datetime import timedelta\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\n\n\nclass AuthConfig(BaseSettings):\n    model_config = SettingsConfigDict(env_prefix=\"AUTH_\", env_file=\".env\", extra=\"ignore\")\n\n    JWT_ALG: str\n    JWT_SECRET: str\n    JWT_EXP_MINUTES: int = 5\n    REFRESH_TOKEN_KEY: str\n    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)\n    SECURE_COOKIES: bool = True\n\n\nauth_settings = AuthConfig()\n```\n\n## Dependencies\n\n### Use Annotated, not default-arg `Depends(...)`\n\n`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids\ngotchas with default values.\n\n```python\n# DO — modern Annotated form\nfrom typing import Annotated\nfrom fastapi import Depends\n\nPostDep = Annotated[dict, Depends(valid_post_id)]\n\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: PostDep):\n    return post\n\n# Avoid — default-argument form (still works, but legacy)\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: dict = Depends(valid_post_id)):\n    return post\n```\n\n### Validate inside dependencies (not just inject)\n```python\nasync def valid_post_id(post_id: UUID4) -> dict:\n    post = await service.get_by_id(post_id)\n    if not post:\n        raise PostNotFound()\n    return post\n```\n\n### Chain dependencies for reuse\n```python\nasync def valid_owned_post(\n    post: Annotated[dict, Depends(valid_post_id)],\n    token_data: Annotated[dict, Depends(parse_jwt_data)],\n) -> dict:\n    if post[\"creator_id\"] != token_data[\"user_id\"]:\n        raise UserNotOwner()\n    return post\n```\n\n### Rules\n- Dependencies are **cached per request**. Same `Depends(x)` called 5 times in one request → `x` runs once.\n- Prefer `async def` dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks.\n- Use **the same path-variable name** across endpoints when you want to share a dependency (e.g. `profile_id` in both `/profiles/{profile_id}` and `/creators/{profile_id}`).\n\n## Authentication — JWT\n\nUse **`PyJWT`**, not `python-jose` (unmaintained).\n\n```python\nimport jwt  # PyJWT\nfrom jwt.exceptions import InvalidTokenError\n\ndef decode_token(token: str) -> dict:\n    try:\n        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])\n    except InvalidTokenError as exc:\n        raise InvalidCredentials() from exc\n```\n\n## Database — SQLAlchemy 2.0 async\n\nPrefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.\n\n```python\n# src/database.py\nfrom sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine\n\nengine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)\nSessionFactory = async_sessionmaker(engine, expire_on_commit=False)\n\n\nasync def get_db() -> AsyncSession:\n    async with SessionFactory() as session:\n        yield session\n```\n\n### Naming conventions\n- `lower_case_snake`\n- Singular tables: `post`, `user`, `post_like`\n- Group with prefix: `payment_account`, `payment_bill`\n- `_at` suffix for `datetime`, `_date` suffix for `date`\n- Use the same FK column name everywhere it appears (`profile_id`, not `user_id` in some tables and `profile_id` in others)\n\n### Index naming convention\n```python\nfrom sqlalchemy import MetaData\n\nPOSTGRES_INDEXES_NAMING_CONVENTION = {\n    \"ix\": \"%(column_0_label)s_idx\",\n    \"uq\": \"%(table_name)s_%(column_0_name)s_key\",\n    \"ck\": \"%(table_name)s_%(constraint_name)s_check\",\n    \"fk\": \"%(table_name)s_%(column_0_name)s_fkey\",\n    \"pk\": \"%(table_name)s_pkey\",\n}\nmetadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)\n```\n\n### SQL-first, Pydantic-second\n- Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this.\n- Hydrate the result into Pydantic only for response validation, not for transformation.\n\n## Background work — BackgroundTasks vs Celery\n\n| Use BackgroundTasks when…                | Use Celery / Arq / RQ when…                |\n|------------------------------------------|--------------------------------------------|\n| Task is < 1 second                       | Task takes seconds to minutes              |\n| Failure can be silently dropped          | You need retries, dead-letter, or visibility|\n| Task is in-process (send email, log row) | Task is CPU-heavy or needs a separate pool |\n| You don't need scheduling                | You need cron, ETA, or rate limiting       |\n\n```python\nfrom fastapi import BackgroundTasks\n\n@router.post(\"/signup\")\nasync def signup(data: SignupIn, bg: BackgroundTasks):\n    user = await service.create_user(data)\n    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process\n    return user\n```\n\n> BackgroundTasks run **after the response is sent, in the same worker process**. If the\n> worker dies, the task is lost. There is no retry. Don't use them for anything you'd\n> page on.\n\n## Testing\n\n### Async client from day one\n```python\nimport pytest\nfrom httpx import AsyncClient, ASGITransport\n\nfrom src.main import app\n\n\n@pytest.fixture\nasync def client():\n    transport = ASGITransport(app=app)\n    async with AsyncClient(transport=transport, base_url=\"http://test\") as ac:\n        yield ac\n\n\n@pytest.mark.asyncio\nasync def test_create_post(client: AsyncClient):\n    resp = await client.post(\"/posts\", json={\"title\": \"hi\"})\n    assert resp.status_code == 201\n```\n\n> **Don't** use `async_asgi_testclient` — it's unmaintained. The example above (httpx +\n> `ASGITransport`) is the supported path.\n\n### Override dependencies in tests\nDon't monkeypatch internals. Use FastAPI's built-in `dependency_overrides`.\n\n```python\nfrom src.auth.dependencies import parse_jwt_data\nfrom src.main import app\n\n\ndef fake_user():\n    return {\"user_id\": \"00000000-0000-0000-0000-000000000001\"}\n\n\n@pytest.fixture(autouse=True)\ndef _override_auth():\n    app.dependency_overrides[parse_jwt_data] = fake_user\n    yield\n    app.dependency_overrides.clear()\n```\n\n## Migrations (Alembic)\n\n- Migrations must be static and reversible.\n- Use the async template: `alembic init -t async migrations`\n- Descriptive filenames:\n  ```ini\n  # alembic.ini\n  file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s\n  ```\n  → `2026-04-14_add_post_content_idx.py`\n\n## API documentation\n\n### Hide docs outside selected envs\n```python\nfrom fastapi import FastAPI\nfrom src.config import settings\n\nSHOW_DOCS_IN = {\"local\", \"staging\"}\napp_kwargs = {\"title\": \"My API\"}\nif settings.ENVIRONMENT not in SHOW_DOCS_IN:\n    app_kwargs[\"openapi_url\"] = None    # disables /docs and /redoc\n\napp = FastAPI(**app_kwargs)\n```\n\n### Document endpoints fully\n```python\nfrom fastapi import APIRouter, status\n\nrouter = APIRouter()\n\n\n@router.post(\n    \"/items\",\n    response_model=ItemResponse,\n    status_code=status.HTTP_201_CREATED,\n    summary=\"Create an item\",\n    description=\"Creates an item owned by the authenticated user.\",\n    tags=[\"items\"],\n    responses={\n        status.HTTP_400_BAD_REQUEST: {\"model\": ErrorResponse, \"description\": \"Validation error\"},\n        status.HTTP_409_CONFLICT:    {\"model\": ErrorResponse, \"description\": \"Slug already exists\"},\n    },\n)\nasync def create_item(payload: ItemCreate) -> ItemResponse: ...\n```\n\n## Linting\n\n```shell\nruff check --fix src\nruff format src\n```\n\nAdd to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.\n\n---\n\n## Anti-patterns — common AI-agent mistakes\n\nIf you're an agent reviewing a diff, check for these. Each is a real failure mode I've\nseen agents introduce.\n\n| Anti-pattern | Why it's wrong | Fix |\n|---|---|---|\n| `requests.get(...)` inside `async def` | Blocks the event loop. `requests` is sync. | Use `httpx.AsyncClient` or `await run_in_threadpool(requests.get, ...)`. |\n| `time.sleep` / `open()` / sync DB driver inside `async def` | Same — blocks the loop. | Use the async equivalent (`asyncio.sleep`, `aiofiles`, async driver). |\n| `from jose import jwt` | `python-jose` is unmaintained. | `import jwt` (PyJWT). |\n| `from async_asgi_testclient import TestClient` | Unmaintained. | `httpx.AsyncClient` + `ASGITransport`. |\n| `model_config = ConfigDict(json_encoders={...})` | Deprecated in Pydantic v2. | `@field_serializer` or `Annotated[T, PlainSerializer(...)]`. |\n| `Field(ge=18, default=None)` | Constraint contradicts the default. | Pick required or optional, not both. |\n| `def get_user(id: int = Depends(...))` (default-arg form) | Legacy; gotchas with default values. | `user: Annotated[User, Depends(...)]`. |\n| Catching `Exception` around a route's body | Hides bugs and turns 500s into silent 200s. | Catch the specific exception class; raise `HTTPException` with a meaningful status. |\n| `BackgroundTasks` for anything you'd page on | No retry, dies with the worker. | Use Celery / Arq / RQ. |\n| Calling a sync ORM session inside `async def` | Blocks the loop, may deadlock the pool. | Use `AsyncSession`. |\n| Returning a Pydantic model and *also* setting `response_model=` to that same class | Model gets constructed twice (validate + serialize). | Either return a `dict`/ORM row and let `response_model` validate, or drop `response_model` and trust the return type. |\n| Importing across domains via deep paths (`from src.auth.service.user import ...`) | Tight coupling, hard to refactor. | `from src.auth import service as auth_service`. |\n| Reusing one `BaseSettings` for the whole app | Hard to reason about, every domain reads every var. | One `BaseSettings` per domain. |\n| Mocking the database in integration tests | Mock/prod divergence eventually fires in prod. | Use a real DB (testcontainers, ephemeral schema) and `dependency_overrides` for auth/external services. |\n\n## Quick reference\n\n| Scenario                             | Solution                                          |\n|--------------------------------------|---------------------------------------------------|\n| Non-blocking I/O                     | `async def` route with `await`                    |\n| Blocking I/O (no async client)       | `def` route (sync, runs in threadpool)            |\n| Sync library inside async route      | `await run_in_threadpool(fn, *args)`              |\n| CPU-intensive work                   | Celery / Arq / RQ worker process                  |\n| Request validation against DB        | Dependency that loads + validates + returns       |\n| Reuse validation across routes       | Chain dependencies                                |\n| Inject dependency in modern style    | `Annotated[T, Depends(...)]`                      |\n| Per-request dep caching              | Default behavior — same `Depends(x)` runs once    |\n| Per-domain config                    | One `BaseSettings` subclass per domain            |\n| Custom datetime serialization        | `@field_serializer`                               |\n| Fire-and-forget short task           | `BackgroundTasks`                                 |\n| Reliable / scheduled / heavy task    | Celery / Arq / RQ                                 |\n| JWT decode                           | `PyJWT` (`import jwt`)                            |\n| Async DB                             | SQLAlchemy 2.0 async (`AsyncSession`)             |\n| HTTP test client                     | `httpx.AsyncClient` + `ASGITransport`             |\n| Swap dep in tests                    | `app.dependency_overrides[dep] = fake`            |\n| Lint + format                        | `ruff check --fix` + `ruff format`                |\n","isInternal":false,"tokens":4108,"sizeBytes":17116}],"systemPromptSnippet":"<yakaai_skills repo=\"zhanymkanov/fastapi-best-practices\">\n<!-- File: AGENTS.md (Tokens: ~4108 | Category: root-instruction) -->\n# FastAPI Best Practices for AI Agents\n\nA machine-readable companion to [README.md](./README.md) for AI coding agents\nworking in FastAPI projects. Same rules, restructured for fast pattern matching:\nversion pins, Do/Don't blocks, anti-patterns, and a quick-reference table.\n\n## Compatibility Matrix\n\nPin to these versions or newer. Examples in this file assume them.\n\n| Dependency       | Minimum   | Notes                                                |\n|------------------|-----------|------------------------------------------------------|\n| Python           | 3.11      | Required for `StrEnum` and `X \\| Y` union syntax     |\n| FastAPI          | 0.115     | `Annotated[T, Depends(...)]` is the idiomatic form   |\n| Pydantic         | 2.7       | v1 APIs (`json_encoders`, `.dict()`) are removed     |\n| pydantic-settings| 2.4       | Lives in a separate package since Pydantic v2        |\n| SQLAlchemy       | 2.0       | Use the async API (`AsyncSession`, `async_sessionmaker`) |\n| Alembic          | 1.13      | Async-aware migrations                               |\n| httpx            | 0.27      | Use `ASGITransport` for in-process tests             |\n| PyJWT            | 2.9       | Use this, not the unmaintained `python-jose`         |\n| ruff             | 0.6       | Replaces black, isort, autoflake                     |\n\n## Project Structure\n\nOrganize by domain, not by file type. One package per bounded context.\n\n```\nsrc/\n├── {domain}/           # e.g., auth/, posts/, aws/\n│   ├── router.py       # API endpoints\n│   ├── schemas.py      # Pydantic models\n│   ├── models.py       # SQLAlchemy ORM models\n│   ├── service.py      # Business logic\n│   ├── dependencies.py # Route dependencies\n│   ├── config.py       # Domain-scoped BaseSettings\n│   ├── constants.py    # Constants and error codes\n│   ├── exceptions.py   # Domain-specific exceptions\n│   └── utils.py        # Helper functions\n├── config.py           # Global BaseSettings\n├── models.py           # Shared Pydantic / ORM bases\n├── exceptions.py       # Global exceptions\n├── database.py         # Async engine + session factory\n└── main.py             # FastAPI app + lifespan\n```\n\n**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`.\n\n```python\nfrom src.auth import constants as auth_constants\nfrom src.notifications import service as notification_service\nfrom src.posts.constants import ErrorCode as PostsErrorCode\n```\n\n## Async Routes\n\n### Decision rule\n\n| Route does this                        | Use         |\n|----------------------------------------|-------------|\n| `await`-able non-blocking I/O          | `async def` |\n| Blocking I/O (no async client exists)  | `def` (sync, runs in threadpool) |\n| Mix of both                            | `async def` + `run_in_threadpool` for the blocking part |\n| CPU-bound work (>50 ms compute)        | Offload to a worker process (Celery / RQ / Arq) |\n\n### Do / Don't\n\n```python\n# DON'T — blocking call inside async route freezes the entire event loop\n@router.get(\"/bad\")\nasync def bad():\n    time.sleep(10)            # blocks every request on this worker\n    return {\"ok\": True}\n\n# DO — sync route lets FastAPI run it in a threadpool\n@router.get(\"/sync-ok\")\ndef sync_ok():\n    time.sleep(10)            # blocks one threadpool worker, not the loop\n    return {\"ok\": True}\n\n# DO — async route with awaitable sleep\n@router.get(\"/async-ok\")\nasync def async_ok():\n    await asyncio.sleep(10)   # yields control, loop keeps serving requests\n    return {\"ok\": True}\n\n# DO — async route that has to call a sync library\nfrom fastapi.concurrency import run_in_threadpool\n\n@router.get(\"/wrap\")\nasync def wrap():\n    result = await run_in_threadpool(legacy_sync_client.fetch, \"id\")\n    return result\n```\n\n### Threadpool caveats\n- Default Starlette threadpool size is 40. Saturating it slows every sync route.\n- Threads cost more than coroutines. Don't use sync routes \"just because.\"\n\n## Pydantic\n\n### Use built-in validators\n```python\nfrom enum import StrEnum\nfrom pydantic import AnyUrl, BaseModel, EmailStr, Field\n\n\nclass MusicBand(StrEnum):\n    AEROSMITH = \"AEROSMITH\"\n    QUEEN = \"QUEEN\"\n    ACDC = \"AC/DC\"\n\n\nclass UserCreate(BaseModel):\n    first_name: str = Field(min_length=1, max_length=128)\n    username: str = Field(min_length=1, max_length=128, pattern=r\"^[A-Za-z0-9_-]+$\")\n    email: EmailStr\n    age: int = Field(ge=18)                     # required, must be >= 18\n    favorite_band: MusicBand | None = None\n    website: AnyUrl | None = None\n```\n\n> **Don't** write `Field(ge=18, default=None)`. The constraint and the default contradict\n> each other. Decide: required (`Field(ge=18)`) or optional (`int | None = Field(default=None, ge=18)`).\n\n### Custom base model — modern serialization\n\n`json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` for per-field rules,\nor annotate a custom type with `PlainSerializer`.\n\n```python\nfrom datetime import datetime\nfrom zoneinfo import ZoneInfo\nfrom pydantic import BaseModel, ConfigDict, field_serializer\n\n\nclass CustomModel(BaseModel):\n    model_config = ConfigDict(populate_by_name=True)\n\n    @field_serializer(\"*\", when_used=\"json\", check_fields=False)\n    def _serialize_datetimes(self, value):\n        if isinstance(value, datetime):\n            if value.tzinfo is None:\n                value = value.replace(tzinfo=ZoneInfo(\"UTC\"))\n            return value.strftime(\"%Y-%m-%dT%H:%M:%S%z\")\n        return value\n```\n\n### Split BaseSettings by domain\n\n`pydantic-settings` is its own package since Pydantic v2.\n\n```python\n# src/auth/config.py\nfrom datetime import timedelta\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\n\n\nclass AuthConfig(BaseSettings):\n    model_config = SettingsConfigDict(env_prefix=\"AUTH_\", env_file=\".env\", extra=\"ignore\")\n\n    JWT_ALG: str\n    JWT_SECRET: str\n    JWT_EXP_MINUTES: int = 5\n    REFRESH_TOKEN_KEY: str\n    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)\n    SECURE_COOKIES: bool = True\n\n\nauth_settings = AuthConfig()\n```\n\n## Dependencies\n\n### Use Annotated, not default-arg `Depends(...)`\n\n`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids\ngotchas with default values.\n\n```python\n# DO — modern Annotated form\nfrom typing import Annotated\nfrom fastapi import Depends\n\nPostDep = Annotated[dict, Depends(valid_post_id)]\n\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: PostDep):\n    return post\n\n# Avoid — default-argument form (still works, but legacy)\n@router.get(\"/posts/{post_id}\")\nasync def get_post(post: dict = Depends(valid_post_id)):\n    return post\n```\n\n### Validate inside dependencies (not just inject)\n```python\nasync def valid_post_id(post_id: UUID4) -> dict:\n    post = await service.get_by_id(post_id)\n    if not post:\n        raise PostNotFound()\n    return post\n```\n\n### Chain dependencies for reuse\n```python\nasync def valid_owned_post(\n    post: Annotated[dict, Depends(valid_post_id)],\n    token_data: Annotated[dict, Depends(parse_jwt_data)],\n) -> dict:\n    if post[\"creator_id\"] != token_data[\"user_id\"]:\n        raise UserNotOwner()\n    return post\n```\n\n### Rules\n- Dependencies are **cached per request**. Same `Depends(x)` called 5 times in one request → `x` runs once.\n- Prefer `async def` dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks.\n- Use **the same path-variable name** across endpoints when you want to share a dependency (e.g. `profile_id` in both `/profiles/{profile_id}` and `/creators/{profile_id}`).\n\n## Authentication — JWT\n\nUse **`PyJWT`**, not `python-jose` (unmaintained).\n\n```python\nimport jwt  # PyJWT\nfrom jwt.exceptions import InvalidTokenError\n\ndef decode_token(token: str) -> dict:\n    try:\n        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])\n    except InvalidTokenError as exc:\n        raise InvalidCredentials() from exc\n```\n\n## Database — SQLAlchemy 2.0 async\n\nPrefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.\n\n```python\n# src/database.py\nfrom sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine\n\nengine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)\nSessionFactory = async_sessionmaker(engine, expire_on_commit=False)\n\n\nasync def get_db() -> AsyncSession:\n    async with SessionFactory() as session:\n        yield session\n```\n\n### Naming conventions\n- `lower_case_snake`\n- Singular tables: `post`, `user`, `post_like`\n- Group with prefix: `payment_account`, `payment_bill`\n- `_at` suffix for `datetime`, `_date` suffix for `date`\n- Use the same FK column name everywhere it appears (`profile_id`, not `user_id` in some tables and `profile_id` in others)\n\n### Index naming convention\n```python\nfrom sqlalchemy import MetaData\n\nPOSTGRES_INDEXES_NAMING_CONVENTION = {\n    \"ix\": \"%(column_0_label)s_idx\",\n    \"uq\": \"%(table_name)s_%(column_0_name)s_key\",\n    \"ck\": \"%(table_name)s_%(constraint_name)s_check\",\n    \"fk\": \"%(table_name)s_%(column_0_name)s_fkey\",\n    \"pk\": \"%(table_name)s_pkey\",\n}\nmetadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)\n```\n\n### SQL-first, Pydantic-second\n- Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this.\n- Hydrate the result into Pydantic only for response validation, not for transformation.\n\n## Background work — BackgroundTasks vs Celery\n\n| Use BackgroundTasks when…                | Use Celery / Arq / RQ when…                |\n|------------------------------------------|--------------------------------------------|\n| Task is < 1 second                       | Task takes seconds to minutes              |\n| Failure can be silently dropped          | You need retries, dead-letter, or visibility|\n| Task is in-process (send email, log row) | Task is CPU-heavy or needs a separate pool |\n| You don't need scheduling                | You need cron, ETA, or rate limiting       |\n\n```python\nfrom fastapi import BackgroundTasks\n\n@router.post(\"/signup\")\nasync def signup(data: SignupIn, bg: BackgroundTasks):\n    user = await service.create_user(data)\n    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process\n    return user\n```\n\n> BackgroundTasks run **after the response is sent, in the same worker process**. If the\n> worker dies, the task is lost. There is no retry. Don't use them for anything you'd\n> page on.\n\n## Testing\n\n### Async client from day one\n```python\nimport pytest\nfrom httpx import AsyncClient, ASGITransport\n\nfrom src.main import app\n\n\n@pytest.fixture\nasync def client():\n    transport = ASGITransport(app=app)\n    async with AsyncClient(transport=transport, base_url=\"http://test\") as ac:\n        yield ac\n\n\n@pytest.mark.asyncio\nasync def test_create_post(client: AsyncClient):\n    resp = await client.post(\"/posts\", json={\"title\": \"hi\"})\n    assert resp.status_code == 201\n```\n\n> **Don't** use `async_asgi_testclient` — it's unmaintained. The example above (httpx +\n> `ASGITransport`) is the supported path.\n\n### Override dependencies in tests\nDon't monkeypatch internals. Use FastAPI's built-in `dependency_overrides`.\n\n```python\nfrom src.auth.dependencies import parse_jwt_data\nfrom src.main import app\n\n\ndef fake_user():\n    return {\"user_id\": \"00000000-0000-0000-0000-000000000001\"}\n\n\n@pytest.fixture(autouse=True)\ndef _override_auth():\n    app.dependency_overrides[parse_jwt_data] = fake_user\n    yield\n    app.dependency_overrides.clear()\n```\n\n## Migrations (Alembic)\n\n- Migrations must be static and reversible.\n- Use the async template: `alembic init -t async migrations`\n- Descriptive filenames:\n  ```ini\n  # alembic.ini\n  file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s\n  ```\n  → `2026-04-14_add_post_content_idx.py`\n\n## API documentation\n\n### Hide docs outside selected envs\n```python\nfrom fastapi import FastAPI\nfrom src.config import settings\n\nSHOW_DOCS_IN = {\"local\", \"staging\"}\napp_kwargs = {\"title\": \"My API\"}\nif settings.ENVIRONMENT not in SHOW_DOCS_IN:\n    app_kwargs[\"openapi_url\"] = None    # disables /docs and /redoc\n\napp = FastAPI(**app_kwargs)\n```\n\n### Document endpoints fully\n```python\nfrom fastapi import APIRouter, status\n\nrouter = APIRouter()\n\n\n@router.post(\n    \"/items\",\n    response_model=ItemResponse,\n    status_code=status.HTTP_201_CREATED,\n    summary=\"Create an item\",\n    description=\"Creates an item owned by the authenticated user.\",\n    tags=[\"items\"],\n    responses={\n        status.HTTP_400_BAD_REQUEST: {\"model\": ErrorResponse, \"description\": \"Validation error\"},\n        status.HTTP_409_CONFLICT:    {\"model\": ErrorResponse, \"description\": \"Slug already exists\"},\n    },\n)\nasync def create_item(payload: ItemCreate) -> ItemResponse: ...\n```\n\n## Linting\n\n```shell\nruff check --fix src\nruff format src\n```\n\nAdd to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.\n\n---\n\n## Anti-patterns — common AI-agent mistakes\n\nIf you're an agent reviewing a diff, check for these. Each is a real failure mode I've\nseen agents introduce.\n\n| Anti-pattern | Why it's wrong | Fix |\n|---|---|---|\n| `requests.get(...)` inside `async def` | Blocks the event loop. `requests` is sync. | Use `httpx.AsyncClient` or `await run_in_threadpool(requests.get, ...)`. |\n| `time.sleep` / `open()` / sync DB driver inside `async def` | Same — blocks the loop. | Use the async equivalent (`asyncio.sleep`, `aiofiles`, async driver). |\n| `from jose import jwt` | `python-jose` is unmaintained. | `import jwt` (PyJWT). |\n| `from async_asgi_testclient import TestClient` | Unmaintained. | `httpx.AsyncClient` + `ASGITransport`. |\n| `model_config = ConfigDict(json_encoders={...})` | Deprecated in Pydantic v2. | `@field_serializer` or `Annotated[T, PlainSerializer(...)]`. |\n| `Field(ge=18, default=None)` | Constraint contradicts the default. | Pick required or optional, not both. |\n| `def get_user(id: int = Depends(...))` (default-arg form) | Legacy; gotchas with default values. | `user: Annotated[User, Depends(...)]`. |\n| Catching `Exception` around a route's body | Hides bugs and turns 500s into silent 200s. | Catch the specific exception class; raise `HTTPException` with a meaningful status. |\n| `BackgroundTasks` for anything you'd page on | No retry, dies with the worker. | Use Celery / Arq / RQ. |\n| Calling a sync ORM session inside `async def` | Blocks the loop, may deadlock the pool. | Use `AsyncSession`. |\n| Returning a Pydantic model and *also* setting `response_model=` to that same class | Model gets constructed twice (validate + serialize). | Either return a `dict`/ORM row and let `response_model` validate, or drop `response_model` and trust the return type. |\n| Importing across domains via deep paths (`from src.auth.service.user import ...`) | Tight coupling, hard to refactor. | `from src.auth import service as auth_service`. |\n| Reusing one `BaseSettings` for the whole app | Hard to reason about, every domain reads every var. | One `BaseSettings` per domain. |\n| Mocking the database in integration tests | Mock/prod divergence eventually fires in prod. | Use a real DB (testcontainers, ephemeral schema) and `dependency_overrides` for auth/external services. |\n\n## Quick reference\n\n| Scenario                             | Solution                                          |\n|--------------------------------------|---------------------------------------------------|\n| Non-blocking I/O                     | `async def` route with `await`                    |\n| Blocking I/O (no async client)       | `def` route (sync, runs in threadpool)            |\n| Sync library inside async route      | `await run_in_threadpool(fn, *args)`              |\n| CPU-intensive work                   | Celery / Arq / RQ worker process                  |\n| Request validation against DB        | Dependency that loads + validates + returns       |\n| Reuse validation across routes       | Chain dependencies                                |\n| Inject dependency in modern style    | `Annotated[T, Depends(...)]`                      |\n| Per-request dep caching              | Default behavior — same `Depends(x)` runs once    |\n| Per-domain config                    | One `BaseSettings` subclass per domain            |\n| Custom datetime serialization        | `@field_serializer`                               |\n| Fire-and-forget short task           | `BackgroundTasks`                                 |\n| Reliable / scheduled / heavy task    | Celery / Arq / RQ                                 |\n| JWT decode                           | `PyJWT` (`import jwt`)                            |\n| Async DB                             | SQLAlchemy 2.0 async (`AsyncSession`)             |\n| HTTP test client                     | `httpx.AsyncClient` + `ASGITransport`             |\n| Swap dep in tests                    | `app.dependency_overrides[dep] = fake`            |\n| Lint + format                        | `ruff check --fix` + `ruff format`                |\n</yakaai_skills>"}