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

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. What makes FastAPI so compelling that companies are rewriting their entire API layers around it? And more importantly, how can you harness its power to build robust, production-ready REST APIs from scratch?

If you have spent any time in the Python web ecosystem, you know the landscape has been dominated by two heavyweights for over a decade: Flask, the minimalist micro-framework loved for its simplicity, and Django with its REST Framework, the batteries-included monolith favored by enterprises. Both are excellent tools. But they were designed 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 way to build software.

FastAPI was born into a different world. It leverages modern Python features — type annotations, async/await, Pydantic data validation — to deliver something that feels almost magical: you write plain, annotated Python functions, and the framework automatically generates interactive API documentation, validates every request and response, and runs with performance that rivals Node.js and Go. That is not marketing hype. Independent benchmarks consistently show FastAPI handling 2-5x more requests per second than Flask.

In this guide, we are going to build a complete REST API from zero to deployment. By the end, you will have 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 — you can follow along step by step and have a working API by the time you finish reading.

Let us get started.

Why FastAPI?

Before we write a single line of code, let us understand what makes FastAPI different and why it has taken the Python community by storm.

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. You do not need to install any plugins, write any YAML files, or maintain separate documentation. Your code is your documentation, and it is always in sync.

Type Hints and Pydantic Validation

FastAPI is built on top of Pydantic, the most popular data validation library in Python. You define your request and response models as simple Python classes with type annotations, and FastAPI automatically validates incoming data, serializes outgoing data, and generates accurate schema documentation — all from the same model definition.

Async Support Out of the Box

FastAPI natively supports Python’s async/await syntax. This means your API can handle thousands of concurrent connections efficiently without blocking, which is critical for I/O-bound workloads like database queries, external API calls, and file operations. You can also use regular synchronous functions — FastAPI handles both seamlessly.

Performance Close to Node.js and Go

Thanks 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

Let us set up a clean development environment. We will use Python 3.11+ (though 3.9+ works fine) and create an isolated virtual environment for our project.

Verify Your Python Installation

python3 --version
# Python 3.11.x or higher recommended

Create Your Project Directory

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

Set Up a Virtual Environment

You have two good options here. The classic venv approach:

# 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

Let us set up a clean project structure that will scale as our API grows:

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

Your First API — Hello World

Every journey begins with a single step. Let us create the simplest possible FastAPI application and see the magic in action.

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 it. Seven lines of actual code and you have a working API with two endpoints. Let us run it:

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

The --reload flag enables hot reloading, so the server restarts automatically when you change your code. You should see output like this:

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

Now open your browser and navigate to http://localhost:8000/docs. You will see a beautiful, interactive API documentation page — generated entirely from your code. You can click on any endpoint, hit “Try it out”, and execute requests directly from the browser.

Also check out http://localhost:8000/redoc for an alternative documentation layout, and http://localhost:8000/openapi.json for the raw OpenAPI schema that can be imported into Postman, Insomnia, or any API client.

Building a Complete CRUD API — Task Manager

Now let us build something real. We will create a full task management API with all CRUD operations, proper validation, error handling, and correct HTTP status codes. We will start with in-memory storage to focus on the API design, then add a database later.

Define Pydantic Models

First, let us define our 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

Notice the separation of concerns: TaskCreate is what clients send when creating a task, TaskUpdate allows partial updates (all fields optional), and TaskResponse is what the API returns. This is a critical design pattern — never expose your internal data model directly.

Build the CRUD Endpoints

Now let us build the actual API. 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]

Let us break down the key design decisions in this code:

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, we raise an HTTPException 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 we only update fields that the client explicitly sent. This is the correct behavior for a PUT/PATCH endpoint.

Testing Your CRUD API

Start the server with uvicorn app.main:app --reload and try these requests 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. Let us explore what Pydantic can do beyond the basics we have already seen.

Field Validation

Pydantic’s Field function gives you 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 get clear, actionable error messages without you writing any error handling code.

Path Parameters, Query Parameters, and Request Body

FastAPI provides elegant ways to extract data from every part of an HTTP request. Let us explore each one.

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 great for 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 comes from based on its type: simple types are path or query parameters, while Pydantic models are request body. The Path, Query, and Body functions let you add validation and documentation to each.

Adding a Database with SQLAlchemy

In-memory storage is fine for prototyping, but any real application needs persistent data storage. Let us integrate 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

Now update app/main.py 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 here 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 done, even if an error occurs. This is a clean, testable pattern that avoids global state.

Authentication and Security

No production API is complete without authentication. Let us implement two approaches: 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

Now you can protect any endpoint 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 your API grows, you will need cross-cutting concerns like CORS support (so frontends can call your API), request logging, and global error handling.

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 Your API

FastAPI makes testing exceptionally easy with its built-in TestClient, which is a wrapper around httpx. You can test your entire API without starting a server.

Setting Up Tests

Install pytest if you have not already:

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

Let us take your API from development to production.

Running in Production with Gunicorn

In production, you should run Uvicorn 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 good rule of thumb for the number of workers is (2 x CPU cores) + 1. For a 2-core server, use 5 workers.

Docker Containerization

Create a Dockerfile:

# 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

For cloud deployment, you have several excellent options depending on your scale and budget:

• AWS Lightsail or EC2 — full control, good for small to medium deployments
• Google Cloud Run — serverless containers, scales to zero, pay per request
• Railway or Render — simple PaaS options with generous free tiers
• AWS Lambda with Mangum — serverless deployment using the Mangum ASGI adapter

Best Practices

As your API grows beyond a simple tutorial, these practices will keep your codebase maintainable and your API reliable.

Project Structure for Larger Applications

For larger apps, organize your code 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 would look like this:

# 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)

And in your main file, include 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):
    ...

Conclusion

We have covered a lot of ground in this guide. Starting from a simple “Hello World” endpoint, we built a complete task management API 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 setup with Docker.

What makes FastAPI special is not just any single feature — it is how all these features work together seamlessly. Type hints drive validation, documentation, and editor support simultaneously. Dependency injection keeps your code testable and modular. Pydantic models serve as your single source of truth for data contracts. And the async foundation means your API can handle serious traffic without complex optimization.

Here is a summary of what we built:

If you are coming from Flask, the transition to FastAPI is remarkably smooth — most concepts map directly, and you gain type safety, auto-docs, and performance for free. If you are coming from Django REST Framework, you will appreciate the lighter weight and more explicit architecture while retaining the same level of functionality.

The Python web ecosystem has evolved significantly, and FastAPI represents the current state of the art. Whether you are building a simple microservice, a complex multi-tenant SaaS, or a high-performance data API, FastAPI gives you the tools to do it cleanly and efficiently.

Start building something real today. Take the task manager we built here, extend it with your own features — tags, due dates, user assignments, notifications — and deploy it. The best 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