FastAPI - Building Modern APIs in Python. A Complete Guide
FastAPI Building Modern
backendFastAPI - Building Modern APIs in Python. A Complete Guide
FastAPI is a modern, high-performance web framework for building APIs with Python 3.7+, based on standard Python type hints. Since its initial release in 2018, it has gained tremendous popularity, becoming one of the most widely adopted tools for creating REST APIs and microservices. In this article, you will learn all the key features of FastAPI and how to build professional APIs from scratch.
Why Is FastAPI Gaining Popularity?#
FastAPI stands out from the competition with several significant advantages:
- Performance - comparable to Node.js and Go, thanks to Starlette and Uvicorn
- Development speed - automatic validation, serialization, and documentation
- Fewer bugs - Python's type system eliminates many issues at the coding stage
- Automatic documentation - Swagger UI and ReDoc generated without additional code
- Standards - full compatibility with OpenAPI and JSON Schema
- Async/await support - native asynchronous programming support
Installation is straightforward:
pip install fastapi uvicorn[standard]
A minimal FastAPI application looks like this:
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="A sample API built with FastAPI",
version="1.0.0"
)
@app.get("/")
async def root():
return {"message": "Welcome to FastAPI!"}
@app.get("/health")
async def health_check():
return {"status": "ok"}
Start the development server:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
Automatic OpenAPI/Swagger Documentation#
One of FastAPI's greatest strengths is automatic API documentation generation. Once the application is running, two interfaces are available:
- Swagger UI - interactive documentation at
/docs - ReDoc - alternative documentation at
/redoc
The documentation is generated based on type hints, docstrings, and decorator parameters:
from fastapi import FastAPI, Query, Path
from enum import Enum
class Category(str, Enum):
electronics = "electronics"
books = "books"
clothing = "clothing"
app = FastAPI()
@app.get(
"/products/{product_id}",
summary="Get product",
description="Returns product details based on ID",
response_description="Product data",
tags=["Products"]
)
async def get_product(
product_id: int = Path(..., title="Product ID", ge=1),
include_reviews: bool = Query(False, description="Include reviews")
):
"""
Retrieve detailed product information:
- **product_id**: unique product ID
- **include_reviews**: optionally include user reviews
"""
return {"product_id": product_id, "include_reviews": include_reviews}
Every endpoint, parameter, and model is automatically visible in the documentation without writing any extra code.
Pydantic Models - Data Validation and Serialization#
FastAPI uses the Pydantic library for input and output data validation. Pydantic models define the shape of data and validation rules:
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime
class Address(BaseModel):
street: str = Field(..., min_length=3, max_length=200)
city: str = Field(..., min_length=2, max_length=100)
zip_code: str = Field(..., pattern=r"^\d{5}(-\d{4})?$")
country: str = "US"
class UserCreate(BaseModel):
username: str = Field(
...,
min_length=3,
max_length=50,
description="Unique username"
)
email: EmailStr
password: str = Field(..., min_length=8)
full_name: Optional[str] = None
age: int = Field(..., ge=18, le=120)
address: Optional[Address] = None
tags: List[str] = []
@validator("password")
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError("Password must contain an uppercase letter")
if not any(c.isdigit() for c in v):
raise ValueError("Password must contain a digit")
return v
class Config:
json_schema_extra = {
"example": {
"username": "john_doe",
"email": "john@example.com",
"password": "MyPassword123",
"full_name": "John Doe",
"age": 30,
"tags": ["developer", "python"]
}
}
class UserResponse(BaseModel):
id: int
username: str
email: EmailStr
full_name: Optional[str]
created_at: datetime
class Config:
from_attributes = True
Using models in endpoints:
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
@app.post(
"/users/",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
tags=["Users"]
)
async def create_user(user: UserCreate):
# Pydantic automatically validates input data
# Invalid data returns HTTP 422 with a description of the issues
db_user = save_to_database(user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse, tags=["Users"])
async def get_user(user_id: int):
user = get_from_database(user_id)
if not user:
raise HTTPException(
status_code=404,
detail=f"User with ID {user_id} not found"
)
return user
Path Parameters, Query Parameters, and Request Bodies#
FastAPI provides a flexible parameter system:
from fastapi import FastAPI, Query, Path, Body
from typing import Optional, List
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
# Path parameter - required
item_id: int = Path(..., title="Item ID", ge=1),
# Query parameters - optional
q: Optional[str] = Query(None, min_length=3, max_length=50),
skip: int = Query(0, ge=0, description="Skip N results"),
limit: int = Query(10, ge=1, le=100, description="Result limit"),
# List values in query string: ?tags=python&tags=fastapi
tags: List[str] = Query([], description="Filter by tags"),
# Sorting
sort_by: Optional[str] = Query(
None,
regex="^(name|price|date)$",
description="Sort field"
)
):
return {
"item_id": item_id,
"q": q,
"skip": skip,
"limit": limit,
"tags": tags,
"sort_by": sort_by
}
@app.put("/items/{item_id}")
async def update_item(
item_id: int = Path(..., ge=1),
item: ItemUpdate = Body(...),
importance: int = Body(1, ge=1, le=5)
):
return {"item_id": item_id, "item": item, "importance": importance}
Async/Await - Asynchronous Programming#
FastAPI natively supports asynchronous programming, allowing it to handle many requests concurrently without blocking:
import httpx
import asyncio
from fastapi import FastAPI
app = FastAPI()
# Async endpoint - does not block the server
@app.get("/async-data")
async def fetch_multiple_sources():
async with httpx.AsyncClient() as client:
# Parallel HTTP requests
tasks = [
client.get("https://api.service1.com/data"),
client.get("https://api.service2.com/data"),
client.get("https://api.service3.com/data"),
]
responses = await asyncio.gather(*tasks)
return {
"service1": responses[0].json(),
"service2": responses[1].json(),
"service3": responses[2].json(),
}
# Synchronous endpoint - runs in a thread pool
@app.get("/sync-data")
def fetch_sync_data():
# Blocking operations (e.g., CPU-intensive)
# FastAPI automatically runs this in a separate thread
result = heavy_computation()
return {"result": result}
Dependency Injection System#
FastAPI includes a built-in dependency injection system that allows for elegant management of shared resources and logic across endpoints:
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional
app = FastAPI()
# Simple dependency
async def common_parameters(
q: Optional[str] = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
# Dependency with sub-dependency
async def get_db():
db = SessionLocal()
try:
yield db # Generator - automatic cleanup
finally:
db.close()
async def get_current_user(
token: str = Header(..., alias="Authorization")
):
user = decode_token(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
async def get_current_active_user(
current_user = Depends(get_current_user)
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# Using dependencies in an endpoint
@app.get("/items/")
async def read_items(
commons: dict = Depends(common_parameters),
db = Depends(get_db),
user = Depends(get_current_active_user)
):
items = db.query(Item).offset(commons["skip"]).limit(commons["limit"]).all()
return items
# Dependencies at the router level
from fastapi import APIRouter
router = APIRouter(
prefix="/admin",
tags=["Admin"],
dependencies=[Depends(get_current_active_user)]
)
@router.get("/stats")
async def admin_stats(db = Depends(get_db)):
return {"total_users": db.query(User).count()}
Authentication with OAuth2 and JWT#
FastAPI has built-in support for OAuth2 with JWT tokens:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str | None = None
def create_access_token(data: dict, expires_delta: timedelta | None = 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)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(username)
if user is None:
raise credentials_exception
return user
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(
data={"sub": user.username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_user)):
return current_user
Database Integration with SQLAlchemy#
FastAPI works seamlessly with SQLAlchemy, both in synchronous and asynchronous modes:
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import FastAPI, Depends, HTTPException
from datetime import datetime
from typing import List
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# SQLAlchemy Model
class UserDB(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, index=True)
email = Column(String(100), unique=True, index=True)
hashed_password = Column(String(200))
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
Base.metadata.create_all(bind=engine)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
app = FastAPI()
# CRUD operations
@app.post("/users/", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
# Check if user exists
db_user = db.query(UserDB).filter(
UserDB.email == user.email
).first()
if db_user:
raise HTTPException(
status_code=400,
detail="Email already registered"
)
# Create user
hashed_password = pwd_context.hash(user.password)
db_user = UserDB(
username=user.username,
email=user.email,
hashed_password=hashed_password
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users/", response_model=List[UserResponse])
def list_users(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db)
):
users = db.query(UserDB).offset(skip).limit(limit).all()
return users
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(UserDB).filter(UserDB.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
db.delete(user)
db.commit()
Background Tasks#
FastAPI allows you to run background tasks without blocking the HTTP response:
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import JSONResponse
import smtplib
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
def send_email(email: str, subject: str, body: str):
"""Send email - background task"""
logger.info(f"Sending email to {email}")
with smtplib.SMTP("smtp.example.com") as server:
server.sendmail("noreply@example.com", email,
f"Subject: {subject}\n\n{body}")
logger.info(f"Email sent to {email}")
def generate_report(report_id: int, params: dict):
"""Generate report - time-consuming operation"""
logger.info(f"Generating report {report_id}")
result = process_data(params)
save_report(report_id, result)
logger.info(f"Report {report_id} ready")
@app.post("/users/register")
async def register_user(
user: UserCreate,
background_tasks: BackgroundTasks
):
# Create user immediately
db_user = create_user_in_db(user)
# Send welcome email in the background
background_tasks.add_task(
send_email,
email=user.email,
subject="Welcome!",
body="Thank you for registering with our service."
)
return {"id": db_user.id, "message": "Registered successfully"}
@app.post("/reports/")
async def create_report(
params: ReportParams,
background_tasks: BackgroundTasks
):
report_id = create_report_entry()
background_tasks.add_task(generate_report, report_id, params.dict())
return {"report_id": report_id, "status": "processing"}
WebSocket - Real-Time Communication#
FastAPI offers built-in WebSocket support for bidirectional communication:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json
app = FastAPI()
class ConnectionManager:
"""WebSocket connection manager"""
def __init__(self):
self.active_connections: dict[str, List[WebSocket]] = {}
async def connect(self, websocket: WebSocket, room: str):
await websocket.accept()
if room not in self.active_connections:
self.active_connections[room] = []
self.active_connections[room].append(websocket)
def disconnect(self, websocket: WebSocket, room: str):
self.active_connections[room].remove(websocket)
async def broadcast(self, message: str, room: str):
for connection in self.active_connections.get(room, []):
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/chat/{room}")
async def websocket_chat(websocket: WebSocket, room: str):
await manager.connect(websocket, room)
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
# Broadcast to everyone in the room
await manager.broadcast(
json.dumps({
"user": message["user"],
"text": message["text"],
"room": room
}),
room
)
except WebSocketDisconnect:
manager.disconnect(websocket, room)
await manager.broadcast(
json.dumps({"system": "A user has left the chat"}),
room
)
@app.websocket("/ws/notifications/{user_id}")
async def websocket_notifications(websocket: WebSocket, user_id: int):
await websocket.accept()
try:
while True:
# Send real-time notifications
notification = await get_next_notification(user_id)
await websocket.send_json(notification)
except WebSocketDisconnect:
pass
Testing with TestClient#
FastAPI provides a TestClient based on httpx for easy API testing:
from fastapi.testclient import TestClient
from main import app
import pytest
client = TestClient(app)
class TestUserEndpoints:
def test_create_user(self):
response = client.post(
"/users/",
json={
"username": "testuser",
"email": "test@example.com",
"password": "TestPassword123",
"age": 25
}
)
assert response.status_code == 201
data = response.json()
assert data["username"] == "testuser"
assert data["email"] == "test@example.com"
assert "id" in data
assert "password" not in data # Password should not be in the response
def test_create_user_invalid_email(self):
response = client.post(
"/users/",
json={
"username": "testuser",
"email": "not-an-email",
"password": "TestPassword123",
"age": 25
}
)
assert response.status_code == 422 # Validation error
def test_get_user_not_found(self):
response = client.get("/users/99999")
assert response.status_code == 404
def test_list_users_pagination(self):
response = client.get("/users/?skip=0&limit=10")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) <= 10
# WebSocket test
def test_websocket_chat():
with client.websocket_connect("/ws/chat/test-room") as websocket:
websocket.send_json({"user": "tester", "text": "Hello!"})
data = websocket.receive_json()
assert data["user"] == "tester"
assert data["text"] == "Hello!"
# Test with mocked dependencies
from unittest.mock import MagicMock
def test_with_mock_db():
mock_db = MagicMock()
mock_db.query.return_value.all.return_value = []
app.dependency_overrides[get_db] = lambda: mock_db
response = client.get("/users/")
assert response.status_code == 200
# Clear override
app.dependency_overrides.clear()
# Test with authentication
def test_protected_endpoint():
# Without token
response = client.get("/users/me")
assert response.status_code == 401
# With valid token
token = create_access_token(data={"sub": "testuser"})
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
Performance Comparison#
FastAPI stands out among other popular frameworks:
| Framework | Language | Req/s (JSON) | Req/s (DB query) | Async | |-----------|----------|-------------|-------------------|-------| | FastAPI | Python | ~32,000 | ~12,000 | Yes | | Flask | Python | ~8,000 | ~4,000 | No* | | Django REST | Python | ~5,000 | ~3,500 | Partial | | Express.js | Node.js | ~35,000 | ~14,000 | Yes | | Gin | Go | ~85,000 | ~35,000 | Yes |
*Flask with gevent can handle async, but not natively.
FastAPI is 3-4x faster than Flask and 5-6x faster than Django REST Framework in standard benchmarks. Compared to Express.js, it achieves comparable performance, which is an impressive result for an interpreted language.
Key advantages of FastAPI:
- Vs Flask - automatic validation, documentation, async, type safety
- Vs Django REST - significantly higher performance, simpler setup, better async
- Vs Express.js - better validation, automatic documentation, type hints
Deployment with Docker and Uvicorn#
Professional FastAPI deployment requires proper ASGI server configuration and containerization:
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY ./app ./app
# Create non-root user
RUN adduser --disabled-password --no-create-home appuser
USER appuser
# Run with Uvicorn
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
Docker Compose configuration with database and cache:
version: "3.8"
services:
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/mydb
- REDIS_URL=redis://redis:6379
- SECRET_KEY=${SECRET_KEY}
depends_on:
db:
condition: service_healthy
redis:
condition: service_started
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
db:
image: postgres:16-alpine
volumes:
- postgres_data:/var/lib/postgresql/data
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
- POSTGRES_DB=mydb
healthcheck:
test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
interval: 5s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
volumes:
postgres_data:
redis_data:
For production environments, using Gunicorn as a process manager is recommended:
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--access-logfile - \
--error-logfile - \
--timeout 120
Production Project Structure#
A well-organized FastAPI project should follow this structure:
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI application
│ ├── config.py # Configuration (env variables)
│ ├── database.py # Database connection
│ ├── dependencies.py # Shared dependencies
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── users.py
│ │ ├── products.py
│ │ └── auth.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py # SQLAlchemy models
│ │ └── product.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py # Pydantic models
│ │ └── product.py
│ ├── services/
│ │ ├── __init__.py
│ │ ├── user_service.py # Business logic
│ │ └── email_service.py
│ └── middleware/
│ ├── __init__.py
│ └── logging.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_users.py
│ └── test_products.py
├── alembic/ # Database migrations
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env
Summary#
FastAPI is an exceptionally mature framework that combines high performance with an outstanding developer experience. With automatic validation, documentation, and typing, the API development process is faster and less error-prone. Native async/await support, the dependency injection system, WebSocket handling, and seamless database integration make it an ideal choice for both simple projects and complex enterprise systems.
Key reasons to choose FastAPI:
- Development speed - less code, automatic documentation
- Performance - comparable to Node.js and Go
- Type safety - validation at every level
- Ecosystem - integration with Pydantic, SQLAlchemy, Celery, and many more
- Production readiness - easy deployment with Docker and Uvicorn
Need a Professional API?#
At MDS Software Solutions Group, we specialize in:
- Designing and building APIs with FastAPI and Python
- Microservices architecture and distributed systems
- Database integration and third-party system connectivity
- Production deployments with Docker and Kubernetes
- Performance optimization and API security audits
Contact us to discuss your project!
Team of programming experts specializing in modern web technologies.