ViggyV FastAPI Builder
You are an expert at building APIs with FastAPI.
Activation
This skill activates when the user needs help with:
- FastAPI application setup
- API endpoint design
- Request/response handling
- Authentication in FastAPI
- Database integration
Process
1. FastAPI Application Structure
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api import router
from app.core.config import settings
from app.db import init_db
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
await init_db()
yield
# Shutdown
pass
app = FastAPI(
title=settings.app_name,
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(router, prefix="/api/v1")
@app.get("/health")
async def health():
return {"status": "healthy"}
2. Pydantic Schemas
# schemas.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field
class UserBase(BaseModel):
email: EmailStr
name: str = Field(..., min_length=1, max_length=100)
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
email: EmailStr | None = None
name: str | None = Field(None, min_length=1, max_length=100)
class UserResponse(UserBase):
id: str
created_at: datetime
is_active: bool
class Config:
from_attributes = True
class PaginatedResponse(BaseModel):
items: list
total: int
page: int
per_page: int
pages: int
3. CRUD Endpoints
# api/users.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.db import get_db
from app.schemas import UserCreate, UserResponse, UserUpdate
from app.services import UserService
router = APIRouter(prefix="/users", tags=["users"])
@router.get("", response_model=list[UserResponse])
async def list_users(
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
service = UserService(db)
return await service.list(skip=skip, limit=limit)
@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
user_id: str,
db: AsyncSession = Depends(get_db),
):
service = UserService(db)
user = await service.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.post("", response_model=UserResponse, status_code=201)
async def create_user(
data: UserCreate,
db: AsyncSession = Depends(get_db),
):
service = UserService(db)
return await service.create(data)
@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
user_id: str,
data: UserUpdate,
db: AsyncSession = Depends(get_db),
):
service = UserService(db)
user = await service.update(user_id, data)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{user_id}", status_code=204)
async def delete_user(
user_id: str,
db: AsyncSession = Depends(get_db),
):
service = UserService(db)
deleted = await service.delete(user_id)
if not deleted:
raise HTTPException(status_code=404, detail="User not found")
4. Authentication
# auth.py
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from app.core.config import settings
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
def create_access_token(user_id: str) -> str:
expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire)
payload = {"sub": user_id, "exp": expire}
return jwt.encode(payload, settings.secret_key, algorithm="HS256")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = await UserService(db).get(user_id)
if user is None:
raise credentials_exception
return user
# Protected endpoint
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: User = Depends(get_current_user)):
return current_user
5. Error Handling
# exceptions.py
from fastapi import Request
from fastapi.responses import JSONResponse
class AppException(Exception):
def __init__(self, status_code: int, detail: str, code: str = None):
self.status_code = status_code
self.detail = detail
self.code = code
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={"error": {"message": exc.detail, "code": exc.code}},
)
# Register in main.py
app.add_exception_handler(AppException, app_exception_handler)
6. Database Setup
# db.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings
engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_db():
async with async_session() as session:
yield session
Output Format
Provide:
- Application setup
- Route definitions
- Schema validation
- Authentication flow
- Database integration