Fastapi Coder

FastAPI Coder

Core Principles

Project Structure

app/
├── main.py              # FastAPI app initialization
├── api/
│   ├── __init__.py
│   ├── deps.py          # Shared dependencies
│   └── routes/          # Route handlers by domain
│       ├── users.py
│       └── items.py
├── core/
│   ├── config.py        # Settings via pydantic-settings
│   ├── security.py      # Auth utilities
│   └── exceptions.py    # Custom exceptions
├── models/              # Pydantic schemas
│   ├── user.py
│   └── item.py
├── services/            # Business logic
│   └── user_service.py
├── repositories/        # Data access
│   └── user_repo.py
└── tests/
    ├── conftest.py      # Shared fixtures
    └── test_users.py

Essential Patterns

Route Handler

from fastapi import APIRouter, Depends, HTTPException, status
from app.models.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.api.deps import get_user_service

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    """Create a new user."""
    return await service.create(user_in)

Pydantic Models

from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

Dependencies

from typing import Annotated
from fastapi import Depends, Header, HTTPException
from app.core.security import verify_token

async def get_current_user(
    authorization: Annotated[str, Header()],
) -> User:
    token = authorization.removeprefix("Bearer ")
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

Service Layer

from app.repositories.user_repo import UserRepository
from app.models.user import UserCreate, UserResponse

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create(self, user_in: UserCreate) -> UserResponse:
        # Business logic here
        existing = await self.repo.get_by_email(user_in.email)
        if existing:
            raise ValueError("Email already registered")
        return await self.repo.create(user_in)

Exception Handling

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

Background Tasks

from fastapi import BackgroundTasks

async def send_welcome_email(email: str):
    # Async email sending
    ...

@router.post("/users/")
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
):
    user = await create_user_in_db(user_in)
    background_tasks.add_task(send_welcome_email, user.email)
    return user

Database Integration

SQLAlchemy 2.0 Async

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://...", echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

Repository Pattern

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

Authentication Patterns

JWT Authentication

from datetime import datetime, timedelta
from jose import jwt, JWTError
from app.core.config import settings

def create_access_token(data: dict) -> str:
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({**data, "exp": expire}, settings.SECRET_KEY, algorithm="HS256")

async def verify_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

Testing Patterns

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

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

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={
        "email": "test@example.com",
        "name": "Test User",
        "password": "securepass123",
    })
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

Quality Checklist

• All routes have response_model and status_code
• Pydantic models for all request/response data
• Dependencies for shared logic (auth, db, services)
• Service layer separates business logic from routes
• Repository pattern for data access
• Custom exceptions with proper handlers
• Async database operations
• Background tasks for non-blocking operations
• Comprehensive tests with httpx AsyncClient
• Type hints throughout