🇺🇸English
FastAPI Expert
Expert guidance for FastAPI - modern, fast Python web framework for building APIs with automatic OpenAPI documentation and type safety.
Core Concepts
FastAPI Features
- Fast performance (Starlette + Pydantic)
- Automatic OpenAPI/Swagger docs
- Type hints and validation
- Async/await support
- Dependency injection
- OAuth2 and JWT
- WebSocket support
Key Components
- Path operations (routes)
- Request/response models (Pydantic)
- Dependency injection
- Middleware
- Background tasks
- Testing with TestClient
Basic FastAPI Application
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional
import uvicorn
app = FastAPI(
title="My API",
description="Production-ready FastAPI",
version="1.0.0"
)
# Models
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
from_attributes = True
# Routes
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# Create user in database
db_user = await db.users.create(**user.dict())
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = await db.users.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 100):
return await db.users.find_many(skip=skip, limit=limit)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Dependency Injection
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
# Database dependency
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
# Auth dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
# Use dependencies
@app.get("/me", response_model=UserResponse)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
@app.post("/posts")
async def create_post(
post: PostCreate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
db_post = Post(**post.dict(), user_id=current_user.id)
db.add(db_post)
await db.commit()
return db_post
Authentication
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
@app.post("/token")
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
):
user = await authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
Database Integration (SQLAlchemy)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, Boolean
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
# CRUD operations
async def get_user(db: AsyncSession, user_id: int):
return await db.get(User, user_id)
async def create_user(db: AsyncSession, user: UserCreate):
hashed_password = get_password_hash(user.password)
db_user = User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
Request Validation
from pydantic import BaseModel, validator, root_validator
from typing import Optional
from datetime import date
class PostCreate(BaseModel):
title: str = Field(..., min_length=5, max_length=200)
content: str = Field(..., min_length=10)
published: bool = False
tags: List[str] = Field(default_factory=list, max_items=10)
@validator('tags')
def validate_tags(cls, v):
return [tag.lower().strip() for tag in v]
@validator('title')
def validate_title(cls, v):
if any(word in v.lower() for word in ['spam', 'xxx']):
raise ValueError('Invalid content')
return v
Background Tasks
from fastapi import BackgroundTasks
def send_email(email: str, message: str):
# Send email
print(f"Sending email to {email}: {message}")
@app.post("/send-notification")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(send_email, email, "Welcome!")
return {"message": "Notification sent"}
Testing
from fastapi.testclient import TestClient
import pytest
client = TestClient(app)
def test_create_user():
response = client.post(
"/users",
json={"email": "test@example.com", "name": "Test User"}
)
assert response.status_code == 201
assert response.json()["email"] == "test@example.com"
def test_get_user():
response = client.get("/users/1")
assert response.status_code == 200
assert "email" in response.json()
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/users")
assert response.status_code == 200
Best Practices
- Use Pydantic models for validation
- Implement proper error handling
- Use dependency injection
- Add rate limiting
- Enable CORS properly
- Use async/await for I/O
- Document with docstrings
- Add proper logging
Resources
Weekly Installs
78
Repository
personamanagmen…ayer/pcl
GitHub Stars
11
First Seen
Jan 24, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode71
codex69
gemini-cli67
github-copilot63
cursor61
claude-code60
