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

# 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

# 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

# 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)

# 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