FastAPI Application Skill

Overview

Expert guidance for building FastAPI backend applications with route decorators, dependency injection, CORS configuration, and Pydantic v2 validation. Supports ERP endpoints for students, fees, attendance, and authentication.

When This Skill Applies

This skill triggers when users request:

• App Setup: "Create FastAPI app", "Initialize FastAPI", "Lifespan events"
• Routes: "Student endpoint", "API route", "GET/POST handler", "APIRouter"
• Dependencies: "DB dependency", "Auth dependency", "Depends()", "JWT auth"
• CORS: "CORS enable frontend", "Cross-origin config", "credentials"
• Models: "Pydantic model", "Student schema", "Fee validation"

Core Rules

1. Init: FastAPI App and Lifespan

python
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import students, fees, attendance, auth
from dependencies.database import get_db
from dependencies.auth import get_current_user
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Starting up FastAPI application...")
    yield
    # Shutdown
    logger.info("Shutting down FastAPI application...")

app = FastAPI(
    title="ERP API",
    description="Educational Resource Planning API",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS Configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers with prefix and tags
app.include_router(
    auth.router,
    prefix="/api/v1/auth",
    tags=["Authentication"],
)

app.include_router(
    students.router,
    prefix="/api/v1/students",
    tags=["Students"],
    dependencies=[Depends(get_current_user)],
)

app.include_router(
    fees.router,
    prefix="/api/v1/fees",
    tags=["Fees"],
    dependencies=[Depends(get_current_user)],
)

app.include_router(
    attendance.router,
    prefix="/api/v1/attendance",
    tags=["Attendance"],
    dependencies=[Depends(get_current_user)],
)

Requirements:

• Use FastAPI() with lifespan context manager
• Configure CORS for frontend origins
• Include routers with APIRouter
• Set up logging for production
• Enable Swagger docs at /docs

2. Routes: APIRouter with Tags and Responses

python
# routers/students.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional
from pydantic import BaseModel, EmailStr, Field

from dependencies.database import get_db
from dependencies.auth import get_current_user, get_admin_user
from models.student import Student as StudentModel
from schemas.student import StudentCreate, StudentUpdate, StudentResponse

router = APIRouter()

@router.get("/", response_model=List[StudentResponse])
async def get_students(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    """Get all students with pagination"""
    students = await db.execute(
        select(StudentModel)
        .offset(skip)
        .limit(limit)
    )
    return students.scalars().all()

@router.get("/{student_id}", response_model=StudentResponse)
async def get_student(
    student_id: str,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    """Get a single student by ID"""
    student = await db.execute(
        select(StudentModel).where(StudentModel.id == student_id)
    )
    student = student.scalar_one_or_none()

    if not student:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Student not found"
        )
    return student

@router.post("/", response_model=StudentResponse, status_code=status.HTTP_201_CREATED)
async def create_student(
    student_data: StudentCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_admin_user),
):
    """Create a new student"""
    # Check if email exists
    existing = await db.execute(
        select(StudentModel).where(StudentModel.email == student_data.email)
    )
    if existing.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )

    student = StudentModel(**student_data.model_dump())
    db.add(student)
    await db.commit()
    await db.refresh(student)
    return student

@router.put("/{student_id}", response_model=StudentResponse)
async def update_student(
    student_id: str,
    student_data: StudentUpdate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_admin_user),
):
    """Update a student"""
    student = await db.execute(
        select(StudentModel).where(StudentModel.id == student_id)
    )
    student = student.scalar_one_or_none()

    if not student:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Student not found"
        )

    update_data = student_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(student, field, value)

    await db.commit()
    await db.refresh(student)
    return student

@router.delete("/{student_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_student(
    student_id: str,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_admin_user),
):
    """Delete a student"""
    student = await db.execute(
        select(StudentModel).where(StudentModel.id == student_id)
    )
    student = student.scalar_one_or_none()

    if not student:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Student not found"
        )

    await db.delete(student)
    await db.commit()

Requirements:

• Use APIRouter with prefix and tags
• Response models with Pydantic schemas
• Proper HTTP status codes (200, 201, 204, 404, 400)
• Dependencies for auth and DB
• Pagination with skip/limit

3. Dependencies: Auth and DB Sessions

python
# dependencies/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from databases import Database
import os

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+asyncpg://user:password@localhost:5432/erp_db"
)

engine = create_async_engine(DATABASE_URL, echo=True)
async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

async def init_db():
    """Initialize database tables"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

python
# dependencies/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional
import os

SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-secret-key")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

security = HTTPBearer()

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    return {"user_id": user_id, "role": payload.get("role")}

async def get_admin_user(current_user = Depends(get_current_user)):
    if current_user.get("role") not in ["admin", "teacher"]:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return current_user

Requirements:

• Async SQLAlchemy sessions with context manager
• JWT token creation and validation
• HTTPBearer for token extraction
• Role-based access control
• Environment variables for secrets

4. Pydantic v2 Models

python
# schemas/student.py
from pydantic import BaseModel, EmailStr, Field, ConfigDict
from datetime import datetime
from typing import Optional
from enum import Enum

class StudentRole(str, Enum):
    STUDENT = "student"
    TEACHER = "teacher"
    ADMIN = "admin"

# Base model with config
class StudentBase(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    phone: Optional[str] = Field(None, pattern=r'^\+?[\d\s-]+$')

# Create schema
class StudentCreate(StudentBase):
    password: str = Field(..., min_length=8)
    class_id: Optional[str] = None

# Update schema (partial updates)
class StudentUpdate(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    name: Optional[str] = Field(None, min_length=2, max_length=100)
    email: Optional[EmailStr] = None
    phone: Optional[str] = None
    class_id: Optional[str] = None

# Response schema
class StudentResponse(StudentBase):
    id: str
    class_id: Optional[str] = None
    created_at: datetime
    updated_at: datetime

# Pagination response
class PaginatedResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    data: list[StudentResponse]
    meta: dict = {
        "total": 0,
        "page": 1,
        "page_size": 100,
        "total_pages": 1
    }

python
# schemas/fees.py
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
from enum import Enum

class FeeStatus(str, Enum):
    PENDING = "pending"
    PAID = "paid"
    OVERDUE = "overdue"
    WAIVED = "waived"

class FeeBase(BaseModel):
    student_id: str
    amount: float = Field(..., gt=0)
    description: str = Field(..., max_length=500)
    due_date: datetime

class FeeCreate(FeeBase):
    pass

class FeeUpdate(BaseModel):
    status: Optional[FeeStatus] = None
    paid_date: Optional[datetime] = None
    notes: Optional[str] = None

class FeeResponse(FeeBase):
    id: str
    status: FeeStatus
    paid_date: Optional[datetime] = None
    created_at: datetime

Requirements:

• Use Pydantic v2 ConfigDict instead of Config class
• Field validation with min/max, patterns, gt/lt
• EmailStr for email validation
• Enum for status fields
• Optional fields with defaults
• From_attributes for ORM compatibility

Output Requirements

Code Files

1. Main Application:

• main.py - FastAPI app initialization
• config.py - Settings and environment variables

1. Routers:

• routers/__init__.py
• routers/students.py
• routers/fees.py
• routers/attendance.py
• routers/auth.py

1. Dependencies:

• dependencies/__init__.py
• dependencies/database.py
• dependencies/auth.py

1. Models and Schemas:

• models/__init__.py
• models/student.py
• schemas/__init__.py
• schemas/student.py
• schemas/fees.py

Integration Requirements

• @api-client: JSON response formatting for frontend
• @auth-integration: JWT validation
• @react-component: Error response schemas

Documentation

• PHR: Create Prompt History Record for auth/DB decisions
• ADR: Document auth strategy (JWT vs session), DB choice (async SQLAlchemy)
• Comments: Document endpoint purposes and validation rules

Workflow

1. Initialize App

• Create FastAPI instance with lifespan
• Configure CORS middleware
• Set up logging

1. Setup Database

• Create async SQLAlchemy engine
• Define session dependency
• Create database models

1. Create Dependencies

• Auth dependency with JWT
• Role-based access
• DB session management

1. Define Schemas

• Pydantic v2 models for requests/responses
• Validation rules
• Response formatting

1. Build Routes

• Create APIRouter for each domain
• Implement CRUD operations
• Add pagination and filtering

1. Test and Document

• Verify Swagger docs
• Test authentication
• Validate error responses

Quality Checklist

Before completing any FastAPI implementation:

• ] Pydantic v2 Validation: ConfigDict, Field validators
• ] SQLAlchemy Async: Use async sessions, avoid blocking calls
• ] Rate Limiting: Implement slowapi or similar for endpoints
• ] Swagger Docs Auto: /docs shows all endpoints
• ] Error Handling: Proper HTTPException with status codes
• ] Auth Protected: All endpoints with dependencies
• ] Pagination: skip/limit for list endpoints
• ] Type Hints: All functions fully typed
• ] Environment Config: Secrets in environment variables
• ] CORS Config: Allow frontend origins with credentials

Common Patterns

Student Endpoint with Auth

python
# routers/students.py
@router.get("/students/", response_model=List[StudentResponse])
async def get_students(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    class_id: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    """Get students with pagination and optional class filter"""
    query = select(StudentModel)

    if class_id:
        query = query.where(StudentModel.class_id == class_id)

    query = query.offset(skip).limit(limit).order_by(StudentModel.created_at.desc())

    result = await db.execute(query)
    return result.scalars().all()

CORS Enable Frontend

python
# main.py
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",   # Next.js dev
        "http://localhost:5173",   # Vite dev
        "https://yourdomain.com",  # Production
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["Authorization", "Content-Type"],
)

DB Dependency with Session

python
# dependencies/database.py
async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

JWT Auth Dependency

python
# dependencies/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer

oauth2_scheme = HTTPBearer()

async def get_token_data(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

Rate Limiting

python
# dependencies/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

def rate_limit(requests: int = 60, seconds: int = 60):
    def decorator(func):
        return limiter.limit(f"{requests}/{seconds}")(func)
    return decorator

# Usage
@router.get("/students/")
@rate_limit(requests=100, seconds=60)
async def get_students():
    return {"message": "Students list"}

Environment Configuration

python
# config.py
from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    APP_NAME: str = "ERP API"
    DEBUG: bool = False
    API_V1_PREFIX: str = "/api/v1"

    # Database
    DATABASE_URL: str = "postgresql+asyncpg://user:password@localhost:5432/erp_db"

    # JWT
    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # CORS
    CORS_ORIGINS: List[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

Running the Application

bash
# Install dependencies
pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg
pip install python-jose[cryptography] passlib[bcrypt]
pip install slowapi pydantic-settings

# Run development
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Run production
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

References

• FastAPI Documentation: https://fastapi.tiangolo.com
• Pydantic v2: https://docs.pydantic.dev/latest/
• SQLAlchemy Async: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
• JWT with FastAPI: https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/
• SlowAPI Rate Limiting: https://pypi.org/project/slowapi/