FastAPI Coder

Core Principles

Principle Application

Async-First Use async/await everywhere, sync only when required

Type Safety Pydantic models for all request/response data

Dependency Injection Use Depends() for shared logic, not global state

OpenAPI-Driven Schema generates automatically; keep it clean

Separation of Concerns Routes → Services → Repositories

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