This post examines FastAPI in 2026 and demonstrates how to construct a production-ready REST API from scratch. In December 2018, a Colombian developer named Sebastián Ramírez pushed the first commit of a Python web framework to GitHub. Six years later, that project — FastAPI — has surpassed 80,000 stars, overtaken Flask in monthly downloads, and become the framework of choice at Netflix, Uber, Microsoft, and hundreds of startups building production APIs. The questions that arise are clear: what makes FastAPI so compelling that companies are rewriting entire API layers around it, and how can its capabilities be applied to build robust, production-ready REST APIs?
For anyone familiar with the Python web ecosystem, the landscape has been dominated by two heavyweights for more than a decade: Flask, the minimalist micro-framework valued for its simplicity, and Django with its REST Framework, the batteries-included monolith favoured by enterprises. Both are excellent tools. They were designed, however, in a world before type hints became standard, before async was a first-class citizen in Python, and before API-first architectures became the default approach to building software.
FastAPI was created in a different environment. It leverages modern Python features that make Python one of the most productive languages available today — type annotations, async/await, and Pydantic data validation — to deliver something that approaches a transformation in developer experience: ordinary annotated Python functions are written, and the framework automatically generates interactive API documentation, validates every request and response, and runs with performance that rivals Node.js and Go. This is not marketing rhetoric. Independent benchmarks consistently show FastAPI handling 2–5x more requests per second than Flask.
This guide builds a complete REST API from zero to deployment. By the end, the reader will possess a fully functional task-management API with CRUD operations, database persistence, authentication, tests, and a production deployment strategy. Every code example is complete and runnable, permitting the reader to follow along step by step and conclude with a working API.
The discussion follows.
Summary
What this post covers: A zero-to-deployment FastAPI tutorial that builds a complete task-manager REST API with CRUD endpoints, Pydantic validation, SQLAlchemy persistence, JWT authentication, tests, and a production deployment strategy.
Key insights:
- FastAPI’s appeal is structural, not cosmetic—type hints + Pydantic + ASGI/Starlette give you automatic OpenAPI docs, request/response validation, and async I/O from the same function signature you would have written anyway.
- Independent benchmarks show FastAPI handling 2–5x more requests per second than Flask, putting it in the same performance class as Node.js and Go for typical I/O-bound workloads.
- Use Pydantic models as the single source of truth for request bodies, response shapes, and OpenAPI schema—if you find yourself duplicating field definitions between models and SQLAlchemy tables, you are doing it wrong.
- Authentication is best implemented with FastAPI’s
Depends()system: a singleget_current_userdependency injected into protected routes keeps JWT decoding, expiry checks, and DB lookups out of your endpoint code. - For production, the right stack is Uvicorn (or Gunicorn with Uvicorn workers) behind Nginx, with structured logging, CORS configured explicitly per origin, and tests written against
TestClientso they exercise the real ASGI app, not a mock.
Main topics: Why FastAPI, Setting Up Your Environment, Your First API—Hello World, Building a Complete CRUD API—Task Manager, Request Validation and Pydantic Models, Path Parameters Query Parameters and Request Body, Adding a Database with SQLAlchemy, Authentication and Security, Middleware CORS and Error Handling, Testing Your API, Deployment, Best Practices.
Why FastAPI?
Before any code is written, the characteristics that distinguish FastAPI and explain its rapid adoption in the Python community warrant examination.
Automatic OpenAPI and Swagger Documentation
Every FastAPI application automatically generates an OpenAPI schema and serves an interactive Swagger UI at /docs and a ReDoc interface at /redoc. No plugins must be installed, no YAML files written, and no separate documentation maintained. The code is the documentation, and the two are always in sync.
Type Hints and Pydantic Validation
FastAPI is built on top of Pydantic, the most widely used data-validation library in Python. Request and response models are defined as simple Python classes with type annotations, and FastAPI automatically validates incoming data, serialises outgoing data, and generates accurate schema documentation — all from the same model definition.
Native Async Support
FastAPI natively supports Python’s async/await syntax. This permits the API to handle thousands of concurrent connections efficiently without blocking, which is critical for I/O-bound workloads such as database queries, external API calls, and file operations. Regular synchronous functions are also supported; FastAPI handles both seamlessly.
Performance Comparable to Node.js and Go
Owing to its ASGI foundation (powered by Starlette) and the Uvicorn server, FastAPI delivers exceptional performance. In the TechEmpower Web Framework Benchmarks, Python ASGI frameworks consistently outperform traditional WSGI frameworks by significant margins.
Framework Comparison
| Feature | FastAPI | Flask | Django REST | Express.js |
|---|---|---|---|---|
| Auto Documentation | Built-in | Plugin required | Plugin required | Plugin required |
| Data Validation | Built-in (Pydantic) | Manual / Marshmallow | Built-in (Serializers) | Manual / Joi |
| Async Support | Native | Limited | Django 4.1+ | Native |
| Performance (req/s) | ~15,000+ | ~3,000 | ~2,500 | ~18,000+ |
| Learning Curve | Easy | Very Easy | Moderate | Easy |
| Type Safety | Full (type hints) | None | Partial | TypeScript optional |
| Dependency Injection | Built-in | No | No | No |
Setting Up Your Environment
A clean development environment is the appropriate starting point. The discussion uses Python 3.11+ (though 3.9+ is also acceptable) and creates an isolated virtual environment for the project.
Verify the Python Installation
python3 --version
# Python 3.11.x or higher recommendedCreate the Project Directory
mkdir fastapi-task-manager
cd fastapi-task-managerSet Up a Virtual Environment
Two options are available. The classic venv approach is one:
# Option 1: Classic venv
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Option 2: Using uv (much faster)
pip install uv
uv venv
source .venv/bin/activateuv should consider trying it. It is a Rust-based Python package manager that installs dependencies 10–100x faster than pip and is rapidly becoming the standard tool for Python project management.
Install FastAPI and Uvicorn
# Install FastAPI with all optional dependencies
pip install "fastapi[standard]"
# This installs:
# - fastapi (the framework)
# - uvicorn (the ASGI server)
# - pydantic (data validation)
# - starlette (the underlying ASGI toolkit)
# - httpx (for testing)
# - python-multipart (for form data)
# - jinja2 (for templates, if needed)Project Structure
A clean project structure that will scale as the API grows is appropriate from the outset:
fastapi-task-manager/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app entry point
│ ├── models.py # Pydantic models (schemas)
│ ├── database.py # Database configuration
│ ├── crud.py # Database operations
│ ├── auth.py # Authentication logic
│ └── routers/
│ ├── __init__.py
│ └── tasks.py # Task endpoints
├── tests/
│ ├── __init__.py
│ └── test_tasks.py # API tests
├── requirements.txt
├── Dockerfile
└── .envCreate the initial directory structure:
mkdir -p app/routers tests
touch app/__init__.py app/routers/__init__.py tests/__init__.pyA First API: Hello World
The most direct illustration begins with the simplest possible FastAPI application. The framework’s behaviour can then be observed.
Create app/main.py:
from fastapi import FastAPI
app = FastAPI(
title="Task Manager API",
description="A complete REST API for managing tasks",
version="1.0.0",
)
@app.get("/")
def read_root():
return {"message": "Welcome to the Task Manager API"}
@app.get("/health")
def health_check():
return {"status": "healthy"}That is the entire requirement. Seven lines of actual code produce a working API with two endpoints. The application is run as follows:
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000The --reload flag enables hot reloading, so the server restarts automatically when code is changed. Output of the following form should appear:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: Started reloader process [12345]
INFO: Started server process [12346]
INFO: Waiting for application startup.
INFO: Application startup complete.Exploring the Swagger UI
Opening a browser at http://localhost:8000/docs reveals an attractive interactive API documentation page, generated entirely from the code. Any endpoint may be clicked, “Try it out” selected, and requests executed directly from the browser.
The alternative documentation layout is available at http://localhost:8000/redoc, and the raw OpenAPI schema — importable into Postman, Insomnia, or any API client — is available at http://localhost:8000/openapi.json.
Building a Complete CRUD API: Task Manager
The following section constructs a substantive example: a full task-management API with all CRUD operations, proper validation, error handling, and correct HTTP status codes. The discussion begins with in-memory storage to focus on API design, and a database is added later.
Define Pydantic Models
The first step is to define the data models. Create app/models.py:
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum
class TaskStatus(str, Enum):
pending = "pending"
in_progress = "in_progress"
completed = "completed"
cancelled = "cancelled"
class TaskCreate(BaseModel):
title: str = Field(
...,
min_length=1,
max_length=200,
description="The title of the task",
examples=["Buy groceries"],
)
description: Optional[str] = Field(
None,
max_length=2000,
description="Detailed description of the task",
)
status: TaskStatus = Field(
default=TaskStatus.pending,
description="Current status of the task",
)
priority: int = Field(
default=1,
ge=1,
le=5,
description="Priority level from 1 (lowest) to 5 (highest)",
)
class TaskUpdate(BaseModel):
title: Optional[str] = Field(
None,
min_length=1,
max_length=200,
)
description: Optional[str] = Field(None, max_length=2000)
status: Optional[TaskStatus] = None
priority: Optional[int] = Field(None, ge=1, le=5)
class TaskResponse(BaseModel):
id: int
title: str
description: Optional[str] = None
status: TaskStatus
priority: int
created_at: datetime
updated_at: datetimeThe separation of concerns is important: TaskCreate represents what clients send when creating a task, TaskUpdate allows partial updates (all fields optional), and TaskResponse represents what the API returns. This is a critical design pattern; the internal data model should never be exposed directly.
Build the CRUD Endpoints
The actual API can now be built. Update app/main.py:
from fastapi import FastAPI, HTTPException, Query
from typing import Optional
from datetime import datetime
from app.models import TaskCreate, TaskUpdate, TaskResponse, TaskStatus
app = FastAPI(
title="Task Manager API",
description="A complete REST API for managing tasks",
version="1.0.0",
)
# In-memory storage
tasks_db: dict[int, dict] = {}
task_id_counter = 0
def get_next_id() -> int:
global task_id_counter
task_id_counter += 1
return task_id_counter
@app.get("/")
def read_root():
return {"message": "Welcome to the Task Manager API"}
@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(
status: Optional[TaskStatus] = Query(
None, description="Filter tasks by status"
),
priority: Optional[int] = Query(
None, ge=1, le=5, description="Filter tasks by priority"
),
skip: int = Query(0, ge=0, description="Number of tasks to skip"),
limit: int = Query(
20, ge=1, le=100, description="Maximum number of tasks to return"
),
):
"""Retrieve all tasks with optional filtering and pagination."""
results = list(tasks_db.values())
# Apply filters
if status is not None:
results = [t for t in results if t["status"] == status]
if priority is not None:
results = [t for t in results if t["priority"] == priority]
# Apply pagination
return results[skip : skip + limit]
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
"""Retrieve a single task by its ID."""
if task_id not in tasks_db:
raise HTTPException(
status_code=404,
detail=f"Task with ID {task_id} not found",
)
return tasks_db[task_id]
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
"""Create a new task."""
now = datetime.utcnow()
task_id = get_next_id()
task_data = {
"id": task_id,
"title": task.title,
"description": task.description,
"status": task.status,
"priority": task.priority,
"created_at": now,
"updated_at": now,
}
tasks_db[task_id] = task_data
return task_data
@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task_update: TaskUpdate):
"""Update an existing task. Only provided fields will be updated."""
if task_id not in tasks_db:
raise HTTPException(
status_code=404,
detail=f"Task with ID {task_id} not found",
)
existing_task = tasks_db[task_id]
update_data = task_update.model_dump(exclude_unset=True)
for field, value in update_data.items():
existing_task[field] = value
existing_task["updated_at"] = datetime.utcnow()
return existing_task
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
"""Delete a task by its ID."""
if task_id not in tasks_db:
raise HTTPException(
status_code=404,
detail=f"Task with ID {task_id} not found",
)
del tasks_db[task_id]The key design decisions in this code merit explanation:
Status code 201 for creation: The POST /tasks endpoint returns 201 (Created) instead of the default 200, which is the correct HTTP semantic for resource creation.
Status code 204 for deletion: The DELETE endpoint returns 204 (No Content) with no response body, which is the standard for successful deletions.
HTTPException for errors: When a task is not found, an HTTPException is raised with a 404 status code and a human-readable detail message. FastAPI converts this into a proper JSON error response automatically.
Partial updates with exclude_unset: The model_dump(exclude_unset=True) call on the update model ensures that only fields explicitly sent by the client are updated. This is the correct behaviour for a PUT/PATCH endpoint.
Testing the CRUD API
The server is started with uvicorn app.main:app --reload, and the following requests may then be issued using curl:
# Create a task
curl -X POST http://localhost:8000/tasks \
-H "Content-Type: application/json" \
-d '{"title": "Learn FastAPI", "description": "Complete the tutorial", "priority": 5}'
# List all tasks
curl http://localhost:8000/tasks
# Get a specific task
curl http://localhost:8000/tasks/1
# Update a task
curl -X PUT http://localhost:8000/tasks/1 \
-H "Content-Type: application/json" \
-d '{"status": "in_progress"}'
# Filter tasks by status
curl "http://localhost:8000/tasks?status=in_progress"
# Delete a task
curl -X DELETE http://localhost:8000/tasks/1http://localhost:8000/docs. It is much faster for exploration than writing curl commands.
Request Validation and Pydantic Models
One of FastAPI’s most powerful features is its deep integration with Pydantic for data validation. The capabilities of Pydantic beyond the basics already discussed are examined below.
Field Validation
Pydantic’s Field function provides fine-grained control over validation:
from pydantic import BaseModel, Field, field_validator
import re
class UserCreate(BaseModel):
username: str = Field(
...,
min_length=3,
max_length=50,
pattern=r"^[a-zA-Z0-9_]+$",
description="Username (letters, numbers, underscores only)",
)
email: str = Field(
...,
min_length=5,
max_length=255,
description="Valid email address",
)
age: int = Field(
...,
gt=0,
lt=150,
description="Age in years",
)
score: float = Field(
default=0.0,
ge=0.0,
le=100.0,
description="Score between 0 and 100",
)
@field_validator("email")
@classmethod
def validate_email(cls, v: str) -> str:
if "@" not in v or "." not in v.split("@")[-1]:
raise ValueError("Invalid email address")
return v.lower()The validation constraints available include:
min_length/max_length— for stringspattern— regex validation for stringsgt/ge/lt/le— greater than, greater or equal, less than, less or equal, for numbersmultiple_of— ensures a number is a multiple of a given value
Nested Models
Pydantic models can be nested to represent complex data structures:
from pydantic import BaseModel
from typing import Optional
class Address(BaseModel):
street: str
city: str
state: str
zip_code: str
country: str = "US"
class ContactInfo(BaseModel):
email: str
phone: Optional[str] = None
address: Optional[Address] = None
class Employee(BaseModel):
name: str
department: str
contact: ContactInfo
tags: list[str] = []
# This would be valid JSON input:
# {
# "name": "Alice",
# "department": "Engineering",
# "contact": {
# "email": "alice@example.com",
# "address": {
# "street": "123 Main St",
# "city": "San Francisco",
# "state": "CA",
# "zip_code": "94102"
# }
# },
# "tags": ["python", "fastapi"]
# }Custom Validators
For complex validation logic that goes beyond simple field constraints, Pydantic offers model validators that can validate relationships between fields:
from pydantic import BaseModel, model_validator
from datetime import date
class DateRange(BaseModel):
start_date: date
end_date: date
@model_validator(mode="after")
def validate_date_range(self):
if self.end_date < self.start_date:
raise ValueError("end_date must be after start_date")
return self
class PasswordChange(BaseModel):
current_password: str
new_password: str = Field(min_length=8)
confirm_password: str
@model_validator(mode="after")
def passwords_match(self):
if self.new_password != self.confirm_password:
raise ValueError("new_password and confirm_password must match")
if self.new_password == self.current_password:
raise ValueError("New password must differ from current password")
return selfWhen validation fails, FastAPI automatically returns a 422 (Unprocessable Entity) response with detailed error messages explaining exactly what went wrong and where. Clients receive clear, actionable error messages without any error-handling code having to be written.
Path Parameters, Query Parameters, and Request Body
FastAPI provides elegant means of extracting data from every part of an HTTP request. Each mechanism is examined below.
Path Parameters
Path parameters are extracted directly from the URL path and are always required:
from fastapi import Path
@app.get("/tasks/{task_id}/comments/{comment_id}")
def get_comment(
task_id: int = Path(..., gt=0, description="The task ID"),
comment_id: int = Path(..., gt=0, description="The comment ID"),
):
return {"task_id": task_id, "comment_id": comment_id}Query Parameters with Pagination
Query parameters are well suited to filtering, sorting, and pagination:
from fastapi import Query
from typing import Optional
from enum import Enum
class SortField(str, Enum):
created_at = "created_at"
priority = "priority"
title = "title"
class SortOrder(str, Enum):
asc = "asc"
desc = "desc"
@app.get("/tasks")
def list_tasks(
# Filtering
status: Optional[TaskStatus] = Query(None),
priority: Optional[int] = Query(None, ge=1, le=5),
search: Optional[str] = Query(
None, min_length=1, max_length=100,
description="Search in title and description",
),
# Sorting
sort_by: SortField = Query(
SortField.created_at, description="Field to sort by"
),
order: SortOrder = Query(
SortOrder.desc, description="Sort order"
),
# Pagination
skip: int = Query(0, ge=0, description="Records to skip"),
limit: int = Query(20, ge=1, le=100, description="Max records"),
):
"""List tasks with filtering, sorting, and pagination."""
results = list(tasks_db.values())
if status:
results = [t for t in results if t["status"] == status]
if priority:
results = [t for t in results if t["priority"] == priority]
if search:
results = [
t for t in results
if search.lower() in t["title"].lower()
or (t["description"] and search.lower() in t["description"].lower())
]
reverse = order == SortOrder.desc
results.sort(key=lambda t: t[sort_by.value], reverse=reverse)
return {
"total": len(results),
"skip": skip,
"limit": limit,
"tasks": results[skip : skip + limit],
}Combining Path, Query, and Body in One Endpoint
from fastapi import Path, Query, Body
@app.put("/projects/{project_id}/tasks/{task_id}")
def update_project_task(
project_id: int = Path(..., gt=0), # From URL path
task_id: int = Path(..., gt=0), # From URL path
notify: bool = Query(False), # From query string
task_update: TaskUpdate = Body(...), # From request body
):
"""
URL: PUT /projects/5/tasks/42?notify=true
Body: {"title": "Updated title", "priority": 3}
"""
# project_id = 5 (from path)
# task_id = 42 (from path)
# notify = True (from query)
# task_update = TaskUpdate(title="Updated title", priority=3) (from body)
return {
"project_id": project_id,
"task_id": task_id,
"notify": notify,
"updates": task_update.model_dump(exclude_unset=True),
}FastAPI automatically determines where each parameter originates based on its type: simple types are path or query parameters, while Pydantic models constitute the request body. The Path, Query, and Body functions allow validation and documentation to be attached to each.
Adding a Database with SQLAlchemy
In-memory storage is acceptable for prototyping, but any real application requires persistent data storage. The following section integrates SQLite with SQLAlchemy; the same pattern works with PostgreSQL, MySQL, or any other database.
Install Database Dependencies
pip install sqlalchemyDatabase Configuration
Create app/database.py:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
# For PostgreSQL:
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}, # SQLite only
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
"""Dependency that provides a database session per request."""
db = SessionLocal()
try:
yield db
finally:
db.close()Define Database Models
Create app/db_models.py:
from sqlalchemy import Column, Integer, String, DateTime, Enum as SQLEnum
from sqlalchemy.sql import func
from app.database import Base
from app.models import TaskStatus
class TaskDB(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
title = Column(String(200), nullable=False)
description = Column(String(2000), nullable=True)
status = Column(
SQLEnum(TaskStatus), default=TaskStatus.pending, nullable=False
)
priority = Column(Integer, default=1, nullable=False)
created_at = Column(
DateTime(timezone=True), server_default=func.now()
)
updated_at = Column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)CRUD Operations Module
Create app/crud.py to separate database logic from endpoint logic:
from sqlalchemy.orm import Session
from typing import Optional
from app.db_models import TaskDB
from app.models import TaskCreate, TaskUpdate, TaskStatus
def get_tasks(
db: Session,
status: Optional[TaskStatus] = None,
priority: Optional[int] = None,
skip: int = 0,
limit: int = 20,
) -> list[TaskDB]:
query = db.query(TaskDB)
if status is not None:
query = query.filter(TaskDB.status == status)
if priority is not None:
query = query.filter(TaskDB.priority == priority)
return query.offset(skip).limit(limit).all()
def get_task(db: Session, task_id: int) -> Optional[TaskDB]:
return db.query(TaskDB).filter(TaskDB.id == task_id).first()
def create_task(db: Session, task: TaskCreate) -> TaskDB:
db_task = TaskDB(**task.model_dump())
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
def update_task(
db: Session, task_id: int, task_update: TaskUpdate
) -> Optional[TaskDB]:
db_task = db.query(TaskDB).filter(TaskDB.id == task_id).first()
if db_task is None:
return None
update_data = task_update.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_task, field, value)
db.commit()
db.refresh(db_task)
return db_task
def delete_task(db: Session, task_id: int) -> bool:
db_task = db.query(TaskDB).filter(TaskDB.id == task_id).first()
if db_task is None:
return False
db.delete(db_task)
db.commit()
return TrueRefactored Endpoints with Database
The endpoints in app/main.py are now updated to use the database:
from fastapi import FastAPI, HTTPException, Query, Depends
from sqlalchemy.orm import Session
from typing import Optional
from app.models import (
TaskCreate, TaskUpdate, TaskResponse, TaskStatus,
)
from app.database import engine, get_db
from app.db_models import Base
from app import crud
# Create database tables on startup
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Task Manager API",
description="A complete REST API for managing tasks",
version="1.0.0",
)
@app.get("/")
def read_root():
return {"message": "Welcome to the Task Manager API"}
@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(
status: Optional[TaskStatus] = Query(None),
priority: Optional[int] = Query(None, ge=1, le=5),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: Session = Depends(get_db),
):
"""Retrieve all tasks with optional filtering and pagination."""
return crud.get_tasks(db, status=status, priority=priority,
skip=skip, limit=limit)
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int, db: Session = Depends(get_db)):
"""Retrieve a single task by its ID."""
task = crud.get_task(db, task_id)
if task is None:
raise HTTPException(status_code=404,
detail=f"Task {task_id} not found")
return task
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
"""Create a new task."""
return crud.create_task(db, task)
@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(
task_id: int,
task_update: TaskUpdate,
db: Session = Depends(get_db),
):
"""Update an existing task."""
task = crud.update_task(db, task_id, task_update)
if task is None:
raise HTTPException(status_code=404,
detail=f"Task {task_id} not found")
return task
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, db: Session = Depends(get_db)):
"""Delete a task by its ID."""
if not crud.delete_task(db, task_id):
raise HTTPException(status_code=404,
detail=f"Task {task_id} not found")The key change is the Depends(get_db) pattern. This is FastAPI’s dependency injection system: it automatically creates a database session for each request and closes it when the request is complete, even if an error occurs. The pattern is clean, testable, and avoids global state.
Authentication and Security
No production API is complete without authentication. Two approaches are implemented below: a simple API key for server-to-server communication, and JWT tokens for user-facing authentication.
Simple API Key Authentication
Create app/auth.py:
from fastapi import Depends, HTTPException, Security, status
from fastapi.security import APIKeyHeader, OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional
from pydantic import BaseModel
# ── API Key Authentication ──────────────────────────
API_KEY = "your-secret-api-key-here" # In production, load from env
api_key_header = APIKeyHeader(name="X-API-Key")
def verify_api_key(api_key: str = Security(api_key_header)):
if api_key != API_KEY:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid API key",
)
return api_key
# ── JWT Authentication ──────────────────────────────
SECRET_KEY = "your-jwt-secret-key" # In production, load from env
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class User(BaseModel):
username: str
email: str
disabled: bool = False
class UserInDB(User):
hashed_password: str
# Simulated user database
fake_users_db = {
"admin": {
"username": "admin",
"email": "admin@example.com",
"hashed_password": pwd_context.hash("secretpassword"),
"disabled": False,
}
}
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def create_access_token(
data: dict, expires_delta: Optional[timedelta] = None
) -> str:
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 get_current_user(token: str = Depends(oauth2_scheme)) -> User:
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_data = fake_users_db.get(username)
if user_data is None:
raise credentials_exception
return User(**user_data)Protecting Endpoints
Any endpoint can now be protected by adding the dependency:
from app.auth import (
verify_api_key, get_current_user, User, Token,
create_access_token, verify_password, fake_users_db,
ACCESS_TOKEN_EXPIRE_MINUTES,
)
from fastapi.security import OAuth2PasswordRequestForm
# Token endpoint for JWT login
@app.post("/token", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends()):
user_data = fake_users_db.get(form_data.username)
if not user_data or not verify_password(
form_data.password, user_data["hashed_password"]
):
raise HTTPException(
status_code=401,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(
data={"sub": form_data.username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
)
return {"access_token": access_token, "token_type": "bearer"}
# Protected endpoint — requires JWT token
@app.get("/users/me", response_model=User)
def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
# Protected endpoint — requires API key
@app.delete("/admin/clear-tasks", dependencies=[Depends(verify_api_key)])
def clear_all_tasks(db: Session = Depends(get_db)):
db.query(TaskDB).delete()
db.commit()
return {"message": "All tasks deleted"}Install the required packages for JWT authentication:
pip install python-jose[cryptography] passlib[bcrypt]SECRET_KEY, API_KEY, and database credentials should always be loaded from environment variables using python-dotenv or pydantic-settings. The hard-coded values here are for tutorial purposes only. For a broader treatment of containerising the API securely, see the related Docker containers explained guide.
Middleware, CORS, and Error Handling
As the API grows, cross-cutting concerns such as CORS support (so that frontends can call the API), request logging, and global error handling become necessary.
Adding CORS for Frontend Access
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000", # React dev server
"https://yourdomain.com", # Production frontend
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)Custom Middleware for Logging and Timing
import time
import logging
from fastapi import Request
logger = logging.getLogger("api")
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
# Process the request
response = await call_next(request)
# Calculate duration
duration = time.time() - start_time
logger.info(
f"{request.method} {request.url.path} "
f"- Status: {response.status_code} "
f"- Duration: {duration:.3f}s"
)
# Add timing header to response
response.headers["X-Process-Time"] = f"{duration:.3f}"
return responseGlobal Exception Handlers
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(
status_code=400,
content={
"error": "Bad Request",
"detail": str(exc),
},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error": "Internal Server Error",
"detail": "An unexpected error occurred",
},
)The general exception handler is particularly important for production: it prevents stack traces from leaking to clients while still logging the full error for debugging.
Testing the API
FastAPI makes testing exceptionally straightforward with its built-in TestClient, which is a wrapper around httpx. The entire API can be tested without starting a server.
Setting Up Tests
Install pytest if it is not already present:
pip install pytest httpxCreate tests/test_tasks.py:
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import Base, get_db
# Use an in-memory SQLite database for tests
TEST_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
)
TestingSessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=engine
)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# Override the database dependency
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
@pytest.fixture(autouse=True)
def setup_database():
"""Create tables before each test, drop after."""
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Welcome to the Task Manager API"}
def test_create_task():
response = client.post(
"/tasks",
json={
"title": "Test Task",
"description": "A test task",
"priority": 3,
},
)
assert response.status_code == 201
data = response.json()
assert data["title"] == "Test Task"
assert data["description"] == "A test task"
assert data["priority"] == 3
assert data["status"] == "pending"
assert "id" in data
assert "created_at" in data
def test_create_task_validation_error():
response = client.post(
"/tasks",
json={"title": "", "priority": 10}, # Empty title, priority too high
)
assert response.status_code == 422
def test_get_task():
# Create a task first
create_response = client.post(
"/tasks", json={"title": "Find me"}
)
task_id = create_response.json()["id"]
# Retrieve it
response = client.get(f"/tasks/{task_id}")
assert response.status_code == 200
assert response.json()["title"] == "Find me"
def test_get_task_not_found():
response = client.get("/tasks/99999")
assert response.status_code == 404
def test_update_task():
# Create a task
create_response = client.post(
"/tasks", json={"title": "Original Title"}
)
task_id = create_response.json()["id"]
# Update it
response = client.put(
f"/tasks/{task_id}",
json={"title": "Updated Title", "status": "in_progress"},
)
assert response.status_code == 200
assert response.json()["title"] == "Updated Title"
assert response.json()["status"] == "in_progress"
def test_delete_task():
# Create a task
create_response = client.post(
"/tasks", json={"title": "Delete me"}
)
task_id = create_response.json()["id"]
# Delete it
response = client.delete(f"/tasks/{task_id}")
assert response.status_code == 204
# Verify it is gone
response = client.get(f"/tasks/{task_id}")
assert response.status_code == 404
def test_list_tasks_with_filter():
# Create tasks with different statuses
client.post(
"/tasks", json={"title": "Task 1", "status": "pending"}
)
client.post(
"/tasks", json={"title": "Task 2", "status": "completed"}
)
client.post(
"/tasks", json={"title": "Task 3", "status": "pending"}
)
# Filter by status
response = client.get("/tasks?status=pending")
assert response.status_code == 200
tasks = response.json()
assert len(tasks) == 2
assert all(t["status"] == "pending" for t in tasks)
def test_list_tasks_pagination():
# Create 5 tasks
for i in range(5):
client.post("/tasks", json={"title": f"Task {i}"})
# Get first page
response = client.get("/tasks?skip=0&limit=2")
assert response.status_code == 200
assert len(response.json()) == 2
# Get second page
response = client.get("/tasks?skip=2&limit=2")
assert response.status_code == 200
assert len(response.json()) == 2Run the tests:
pytest tests/ -vapp.dependency_overrides[get_db] = override_get_db). No mocking, no patching, no test doubles. This is one of FastAPI’s most underappreciated features.
Deployment
The following section describes taking the API from development to production.
Running in Production with Gunicorn
In production, Uvicorn should be run behind Gunicorn for process management and multi-worker support:
pip install gunicorn
# Run with 4 worker processes
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--access-logfile - \
--error-logfile -A useful rule of thumb for the number of workers is (2 x CPU cores) + 1. For a 2-core server, five workers are appropriate.
Docker Containerisation
A Dockerfile is used to containerise the FastAPI application. For a thorough treatment of Docker from development to production, including multi-stage builds and Docker Compose, see the related Docker containers guide for development and production:
# Use the official Python slim image
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Install dependencies first (leverages Docker caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY app/ ./app/
# Create non-root user for security
RUN adduser --disabled-password --gecos "" appuser
USER appuser
# Expose port
EXPOSE 8000
# Run with Gunicorn in production
CMD ["gunicorn", "app.main:app", \
"--workers", "4", \
"--worker-class", "uvicorn.workers.UvicornWorker", \
"--bind", "0.0.0.0:8000"]And a docker-compose.yml for easy local testing:
version: "3.8"
services:
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:password@db:5432/taskmanager
- SECRET_KEY=your-production-secret-key
depends_on:
- db
db:
image: postgres:16
environment:
- POSTGRES_DB=taskmanager
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
volumes:
postgres_data:Build and run:
docker-compose up --buildCloud Deployment Options
Several cloud-deployment options are available, depending on scale and budget:
- AWS Lightsail or EC2 — full control, appropriate for small to medium deployments
- Google Cloud Run — serverless containers, scaling to zero, pay-per-request pricing
- Railway or Render — simple PaaS options with generous free tiers
- AWS Lambda with Mangum — serverless deployment using the Mangum ASGI adapter
Best Practices
As an API grows beyond a simple tutorial, the following practices keep the codebase maintainable and the API reliable.
Project Structure for Larger Applications
For larger applications, the code should be organised using FastAPI’s router system:
app/
├── __init__.py
├── main.py # App factory, middleware, startup events
├── config.py # Settings via pydantic-settings
├── database.py # DB engine, session, base
├── dependencies.py # Shared dependencies (auth, db session)
├── models/ # SQLAlchemy models
│ ├── __init__.py
│ ├── task.py
│ └── user.py
├── schemas/ # Pydantic schemas
│ ├── __init__.py
│ ├── task.py
│ └── user.py
├── routers/ # API route handlers
│ ├── __init__.py
│ ├── tasks.py
│ └── users.py
├── services/ # Business logic
│ ├── __init__.py
│ ├── task_service.py
│ └── user_service.py
└── middleware/ # Custom middleware
├── __init__.py
└── logging.pyEach router file has the following structure:
# app/routers/tasks.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.dependencies import get_db, get_current_user
from app.schemas.task import TaskCreate, TaskResponse
from app.services import task_service
router = APIRouter(
prefix="/tasks",
tags=["tasks"],
dependencies=[Depends(get_current_user)],
)
@router.get("/", response_model=list[TaskResponse])
def list_tasks(db: Session = Depends(get_db)):
return task_service.get_all_tasks(db)The main file then includes the routers:
# app/main.py
from fastapi import FastAPI
from app.routers import tasks, users
app = FastAPI(title="Task Manager API")
app.include_router(tasks.router)
app.include_router(users.router)Environment Variables with Pydantic Settings
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
database_url: str = "sqlite:///./tasks.db"
secret_key: str = "change-me-in-production"
api_key: str = "change-me-in-production"
debug: bool = False
allowed_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
# Usage in endpoints:
# settings = Depends(get_settings)API Versioning
# Version via URL prefix
v1_router = APIRouter(prefix="/api/v1")
v2_router = APIRouter(prefix="/api/v2")
app.include_router(v1_router)
app.include_router(v2_router)Rate Limiting
For rate limiting, the slowapi library integrates cleanly with FastAPI:
pip install slowapifrom slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/tasks")
@limiter.limit("60/minute")
def list_tasks(request: Request):
...Concluding Observations
This guide has covered substantial ground. Beginning from a simple “Hello World” endpoint, a complete task-management API has been constructed with CRUD operations, database persistence using SQLAlchemy, authentication with both API keys and JWT tokens, CORS support, custom middleware, comprehensive tests, and a production deployment configured with Docker.
What distinguishes FastAPI is not any single feature; it is how all of its features work together seamlessly. Type hints drive validation, documentation, and editor support simultaneously. Dependency injection keeps code testable and modular. Pydantic models serve as the single source of truth for data contracts. The async foundation permits the API to handle serious traffic without complex optimisation.
The components constructed in this guide are summarised below:
| Component | Technology | Purpose |
|---|---|---|
| Framework | FastAPI | API routing, validation, docs |
| Server | Uvicorn / Gunicorn | ASGI server for production |
| Validation | Pydantic | Request/response data models |
| Database | SQLAlchemy + SQLite | Persistent data storage |
| Authentication | JWT + API Keys | Secure endpoint access |
| Testing | pytest + TestClient | Automated API testing |
| Deployment | Docker + Gunicorn | Containerized production setup |
For teams seeking still more performance from the API layer, writing performance-critical endpoints as native extensions is becoming practical owing to Python and Rust interoperability via PyO3. For developers migrating from Flask, the transition to FastAPI is remarkably smooth: most concepts map directly, and type safety, auto-generated documentation, and improved performance are gained without additional effort. For developers migrating from Django REST Framework, the lighter weight and more explicit architecture, with comparable functionality, are likely to be appreciated.
The Python web ecosystem has evolved significantly, and FastAPI represents the present state of the art. Whether the project is a simple microservice, a complex multi-tenant SaaS, or a high-performance data API, FastAPI provides the tools to build it cleanly and efficiently.
As the codebase grows, following clean-code principles and using Git best practices for professional developers will keep the API maintainable. Building something real is the appropriate next step. The task manager constructed here can be extended with additional features — tags, due dates, user assignments, notifications — and deployed. The most effective way to learn a framework is to ship something with it.