fastapi-app

What this skill does

# 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, algorith