FastAPI

--- name: FastAPI description: Use when building high-performance Python web APIs with automatic documentation, type hints, and async support. Essential for REST APIs, microservices, and data validation. category: dev metadata: author: skynet version: 1.0.0 --- # FastAPI ## Installation ```bash # Basic installation pip install fastapi uvicorn # With all optional dependencies pip install "fastapi[all]" # For production pip install fastapi uvicorn[standard] gunicorn ``` ## Quick Start ```python # main.py from fastapi import FastAPI app = FastAPI() @app.get("/") def read_root(): return {"Hello": "World"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} ``` ```bash # Development server uvicorn main:app --reload # Production server uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 ``` ## Pydantic Models & Validation ```python from pydantic import BaseModel, EmailStr, validator from datetime import datetime from typing import Optional, List class UserBase(BaseModel): email: EmailStr name: str age: Optional[int] = None class UserCreate(UserBase): password: str @validator('password') def validate_password(cls, v): if len(v) < 8: raise ValueError('Password must be at least 8 characters') return v class UserResponse(UserBase): id: int created_at: datetime class Config: orm_mode = True @app.post("/users/", response_model=UserResponse) def create_user(user: UserCreate): # Process user creation return user ``` ## Path & Query Parameters ```python from enum import Enum from typing import Optional, List class ModelName(str, Enum): alexnet = "alexnet" resnet = "resnet" lenet = "lenet" @app.get("/models/{model_name}") def get_model(model_name: ModelName): return {"model_name": model_name} @app.get("/items/") def read_items( skip: int = 0, limit: int = 100, tags: List[str] = Query(None, description="Tags to filter by"), q: Optional[str] = Query(None, min_length=3, max_length=50) ): return {"skip": skip, "limit": limit, "tags": tags, "q": q} ``` ## Request Body & Forms ```python from fastapi import Form, File, UploadFile # JSON body @app.post("/items/") def create_item(item: Item): return item # Form data @app.post("/login/") def login(username: str = Form(), password: str = Form()): return {"username": username} # File upload @app.post("/files/") def upload_file(file: UploadFile = File(...)): return {"filename": file.filename, "size": file.size} # Multiple files @app.post("/uploadfiles/") def upload_files(files: List[UploadFile] = File(...)): return [{"filename": f.filename} for f in files] ``` ## Dependency Injection ```python from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security = HTTPBearer() def get_db(): db = SessionLocal() try: yield db finally: db.close() def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db) ): token = credentials.credentials user = verify_token(token, db) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials" ) return user @app.get("/protected/") def protected_route(current_user: User = Depends(get_current_user)): return {"user": current_user} ``` ## Middleware & CORS ```python from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware import time # CORS app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Custom middleware @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response ``` ## Error Handling ```python from fastapi import HTTPException from fastapi.responses import JSONResponse class CustomException(Exception): def __init__(self, name: str): self.name = name @app.exception_handler(CustomException) def custom_exception_handler(request: Request, exc: CustomException): return JSONResponse( status_code=418, content={"message": f"Oops! {exc.name} did something wrong."} ) @app.get("/items/{item_id}") def read_item(item_id: int): if item_id == 0: raise HTTPException( status_code=404, detail="Item not found", headers={"X-Error": "Custom error header"} ) return {"item_id": item_id} ``` ## Async Operations ```python import asyncio import httpx @app.get("/async-data") async def get_async_data(): async with httpx.AsyncClient() as client: response = await client.get("https://api.example.com/data") return response.json() @app.post("/process-async/") async def process_async(data: ProcessData): # Simulate async processing await asyncio.sleep(1) result = await process_data_async(data) return {"result": result, "processed_at": datetime.utcnow()} ``` ## Database Integration (SQLAlchemy) ```python from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session # Database setup SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # Models class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) email = Column(String, unique=True, index=True) name = Column(String) # Database operations @app.get("/users/{user_id}", response_model=UserResponse) def get_user(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return user ``` ## Testing ```python # test_main.py from fastapi.testclient import TestClient import pytest client = TestClient(app) def test_read_main(): response = client.get("/") assert response.status_code == 200 assert response.json() == {"Hello": "World"} def test_create_user(): user_data = { "email": "test@example.com", "name": "Test User", "password": "testpassword123" } response = client.post("/users/", json=user_data) assert response.status_code == 200 assert response.json()["email"] == user_data["email"] # Run tests pytest test_main.py -v ``` ## Decision Tree ``` Choose FastAPI approach: ├── Simple CRUD API? → Use Pydantic models + SQLAlchemy ├── High-performance async? → Use async/await with asyncio ├── File uploads? → Use UploadFile with Form data ├── Authentication needed? │ ├── Simple token? → Use HTTPBearer dependency │ └── OAuth2/JWT? → Use FastAPI security utilities ├── External API calls? → Use httpx with async client └── Real-time features? → Consider WebSocket support ``` ## Deployment Commands ```bash # Docker echo "FROM python:3.9 COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD uvicorn main:app --host 0.0.0.0 --port 8000" > Dockerfile docker build -t my-api . docker run -p 8000:8000 my-api # Gunicorn for production gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker # Environment variables export DATABASE_URL="postgresql://user:pass@localhost/db" export SECRET_KEY="your-secret-key" uvicorn main:app --env-file .env ``` ## Troubleshooting **Error: `422 Unprocessable Entity`** ``` Fix: Check Pydantic model validation - Verify required fields are present - Check data types match model definition - Add proper validators for custom validation ``` **Error: `ImportError: No module named 'email_validator'`** ```bash pip install email-validator ``` **Error: `RuntimeError: cannot be called from a running async loop`** ```python # Wrong response = requests.get(url) # Right async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Error: `sqlalchemy.exc.OperationalError: connection failed`** ```python # Check database URL and ensure DB is running SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost:5432/dbname" # Test connection from sqlalchemy import create_engine engine = create_engine(SQLALCHEMY_DATABASE_URL) engine.connect() ```