Building REST APIs with FastAPI: A Modern Python Web Framework Guide

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.

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

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 recommended

Create the Project Directory

mkdir fastapi-task-manager
cd fastapi-task-manager

Set 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/activate

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
└── .env

Create the initial directory structure:

mkdir -p app/routers tests
touch app/__init__.py app/routers/__init__.py tests/__init__.py

A 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 8000

The --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: datetime

The 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/1

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 strings
• pattern — regex validation for strings
• gt / ge / lt / le — greater than, greater or equal, less than, less or equal, for numbers
• multiple_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 self

When 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 sqlalchemy

Database 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 True

Refactored 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]

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 response

Global 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 httpx

Create 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()) == 2

Run the tests:

pytest tests/ -v

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 --build

Cloud 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.py

Each 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 slowapi

from 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:

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.

References

• FastAPI Official Documentation
• Pydantic v2 Documentation
• Uvicorn—ASGI Server
• SQLAlchemy Documentation
• SQLModel—SQL Databases with Pydantic
• TestDriven.io, Developing and Testing a FastAPI CRUD App
• FastAPI Best Practices—GitHub Repository
• TechEmpower Web Framework Benchmarks