Welcome to the definitive learning resource for FastAPI. Whether you're new to web development or a seasoned Python developer, this textbook will guide you through every aspect of building modern APIs.
Chapter 1: Understanding the Foundation
What is FastAPI?
FastAPI is a modern, high-performance web framework for building APIs with Python 3.7+ based on standard Python type hints. Created by Sebastián Ramírez, it has quickly become the most popular Python web framework for new projects.
The Technology Stack
Installation and Setup
# Create a virtual environment
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# Install dependencies
pip install fastapi "uvicorn[standard]" pydantic-settingsChapter 2: HTTP Methods and Routing
The HTTP Protocol
Understanding HTTP is essential for API development:
| Method | Purpose | Idempotent | Safe |
|---|---|---|---|
| GET | Retrieve resources | Yes | Yes |
| POST | Create resources | No | No |
| PUT | Replace resources | Yes | No |
| PATCH | Update resources | No | No |
| DELETE | Remove resources | Yes | No |
RESTful Route Design
from fastapi import FastAPI, HTTPException, status
from typing import List
app = FastAPI()
# Collection endpoints
@app.get("/articles", response_model=List[ArticleResponse])
async def list_articles(skip: int = 0, limit: int = 10):
"""List all articles with pagination."""
pass
@app.post("/articles", response_model=ArticleResponse, status_code=201)
async def create_article(article: ArticleCreate):
"""Create a new article."""
pass
# Item endpoints
@app.get("/articles/{article_id}", response_model=ArticleResponse)
async def get_article(article_id: int):
"""Retrieve a specific article."""
pass
@app.put("/articles/{article_id}", response_model=ArticleResponse)
async def update_article(article_id: int, article: ArticleUpdate):
"""Replace an entire article."""
pass
@app.patch("/articles/{article_id}", response_model=ArticleResponse)
async def partial_update_article(article_id: int, article: ArticlePatch):
"""Partially update an article."""
pass
@app.delete("/articles/{article_id}", status_code=204)
async def delete_article(article_id: int):
"""Delete an article."""
passChapter 3: Data Validation with Pydantic
Schema Design Principles
Pydantic models are the backbone of FastAPI's type safety:
from pydantic import BaseModel, EmailStr, Field, field_validator
from datetime import datetime
from typing import Optional, List
from enum import Enum
class Role(str, Enum):
ADMIN = "admin"
USER = "user"
GUEST = "guest"
class UserBase(BaseModel):
"""Base user model with shared attributes."""
email: EmailStr
username: str = Field(
...,
min_length=3,
max_length=50,
pattern="^[a-zA-Z0-9_]+$"
)
class UserCreate(UserBase):
"""Model for creating a new user."""
password: str = Field(..., min_length=8)
role: Role = Role.USER
@field_validator('password')
@classmethod
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Password must contain uppercase letter')
if not any(c.isdigit() for c in v):
raise ValueError('Password must contain digit')
return v
class UserResponse(UserBase):
"""Model for user responses (excludes password)."""
id: int
role: Role
created_at: datetime
posts_count: int = 0
class Config:
from_attributes = True # Enables ORM modeNested Models
class Address(BaseModel):
street: str
city: str
country: str
postal_code: str
class Company(BaseModel):
name: str
address: Address
employees: List[UserResponse] = []
class CompanyCreate(BaseModel):
name: str
address: AddressChapter 4: Dependency Injection Deep Dive
Understanding Dependencies
Dependencies are reusable components that FastAPI injects into your endpoints:
from fastapi import Depends, Query
from typing import Annotated
# Simple dependency
async def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Parameterized dependency
def pagination_params(
skip: int = Query(0, ge=0, description="Items to skip"),
limit: int = Query(10, ge=1, le=100, description="Items per page")
):
return {"skip": skip, "limit": limit}
# Type alias for cleaner code
PaginationDep = Annotated[dict, Depends(pagination_params)]
DBDep = Annotated[Session, Depends(get_db)]
@app.get("/items")
async def list_items(db: DBDep, pagination: PaginationDep):
return db.query(Item).offset(
pagination["skip"]
).limit(pagination["limit"]).all()Dependency Classes
class AuthorizationChecker:
def __init__(self, required_roles: List[str]):
self.required_roles = required_roles
async def __call__(
self,
current_user: User = Depends(get_current_user)
):
if current_user.role not in self.required_roles:
raise HTTPException(
status_code=403,
detail="Insufficient permissions"
)
return current_user
# Usage
admin_required = AuthorizationChecker(["admin"])
moderator_required = AuthorizationChecker(["admin", "moderator"])
@app.delete("/users/{user_id}")
async def delete_user(
user_id: int,
current_user: User = Depends(admin_required)
):
passChapter 5: Async Programming
The Event Loop
Async programming enables handling thousands of concurrent connections:
import asyncio
from httpx import AsyncClient
# Parallel external API calls
async def fetch_user_data(user_id: int):
async with AsyncClient() as client:
# These run concurrently
user_task = client.get(f"https://api.example.com/users/{user_id}")
posts_task = client.get(f"https://api.example.com/users/{user_id}/posts")
user_response, posts_response = await asyncio.gather(
user_task, posts_task
)
return {
"user": user_response.json(),
"posts": posts_response.json()
}
@app.get("/users/{user_id}/full")
async def get_user_with_posts(user_id: int):
return await fetch_user_data(user_id)When to Use Async
| Operation | Use Async | Reason |
|---|---|---|
| Database queries | Yes | I/O bound |
| External API calls | Yes | Network I/O |
| File operations | Yes | Disk I/O |
| CPU-heavy computation | No | Blocks event loop |
| Image processing | No | Use background task |
Chapter 6: Database Patterns
Repository Pattern
Separate your data access logic:
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, List, Optional
T = TypeVar('T')
class BaseRepository(ABC, Generic[T]):
@abstractmethod
async def get(self, id: int) -> Optional[T]:
pass
@abstractmethod
async def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
pass
@abstractmethod
async def create(self, obj: T) -> T:
pass
@abstractmethod
async def update(self, id: int, obj: T) -> Optional[T]:
pass
@abstractmethod
async def delete(self, id: int) -> bool:
pass
class UserRepository(BaseRepository[User]):
def __init__(self, db: AsyncSession):
self.db = db
async def get(self, id: int) -> Optional[User]:
result = await self.db.execute(
select(User).where(User.id == id)
)
return result.scalar_one_or_none()
async def get_by_email(self, email: str) -> Optional[User]:
result = await self.db.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()Chapter 7: Security Implementation
JWT Authentication Flow
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
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)
async def authenticate_user(db: Session, email: str, password: str):
user = await get_user_by_email(db, email)
if not user or not verify_password(password, user.hashed_password):
return None
return userChapter 8: Testing Strategies
Test Organization
tests/
├── conftest.py # Shared fixtures
├── unit/
│ ├── test_models.py
│ └── test_services.py
├── integration/
│ ├── test_api.py
│ └── test_database.py
└── e2e/
└── test_workflows.pyFixtures and Factories
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
@pytest.fixture
async def async_client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def test_db():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
yield session
@pytest.mark.asyncio
async def test_create_user(async_client, test_db):
response = await async_client.post(
"/users",
json={"email": "test@test.com", "username": "testuser", "password": "Test1234"}
)
assert response.status_code == 201Chapter 9: Deployment
Docker Configuration
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip wheel --no-cache-dir --no-deps --wheel-dir /wheels -r requirements.txt
FROM python:3.11-slim as production
WORKDIR /app
COPY --from=builder /wheels /wheels
RUN pip install --no-cache /wheels/*
COPY ./app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]Kubernetes Manifest
apiVersion: apps/v1
kind: Deployment
metadata:
name: fastapi-app
spec:
replicas: 3
selector:
matchLabels:
app: fastapi
template:
spec:
containers:
- name: api
image: your-registry/fastapi-app:latest
ports:
- containerPort: 8000
livenessProbe:
httpGet:
path: /health
port: 8000
readinessProbe:
httpGet:
path: /ready
port: 8000Conclusion
This textbook has covered the essential concepts needed to build production-grade FastAPI applications. From basic routing to advanced patterns like dependency injection and async programming, you now have the knowledge to architect scalable, maintainable APIs.
Practice makes perfect. The best way to learn is by building. ForgeAPI provides a complete, production-ready codebase that implements all these patterns—perfect for learning by example.