Przejdź do treści
Backend

FastAPI - Building Modern APIs in Python. A Complete Guide

Published on:
·4 min read·Author: MDS Software Solutions Group

FastAPI Building Modern

backend

FastAPI - Building Modern APIs in Python. A Complete Guide

FastAPI is a modern, high-performance web framework for building APIs with Python 3.7+, based on standard Python type hints. Since its initial release in 2018, it has gained tremendous popularity, becoming one of the most widely adopted tools for creating REST APIs and microservices. In this article, you will learn all the key features of FastAPI and how to build professional APIs from scratch.

Why Is FastAPI Gaining Popularity?#

FastAPI stands out from the competition with several significant advantages:

  • Performance - comparable to Node.js and Go, thanks to Starlette and Uvicorn
  • Development speed - automatic validation, serialization, and documentation
  • Fewer bugs - Python's type system eliminates many issues at the coding stage
  • Automatic documentation - Swagger UI and ReDoc generated without additional code
  • Standards - full compatibility with OpenAPI and JSON Schema
  • Async/await support - native asynchronous programming support

Installation is straightforward:

pip install fastapi uvicorn[standard]

A minimal FastAPI application looks like this:

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="A sample API built with FastAPI",
    version="1.0.0"
)

@app.get("/")
async def root():
    return {"message": "Welcome to FastAPI!"}

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Start the development server:

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

Automatic OpenAPI/Swagger Documentation#

One of FastAPI's greatest strengths is automatic API documentation generation. Once the application is running, two interfaces are available:

  • Swagger UI - interactive documentation at /docs
  • ReDoc - alternative documentation at /redoc

The documentation is generated based on type hints, docstrings, and decorator parameters:

from fastapi import FastAPI, Query, Path
from enum import Enum

class Category(str, Enum):
    electronics = "electronics"
    books = "books"
    clothing = "clothing"

app = FastAPI()

@app.get(
    "/products/{product_id}",
    summary="Get product",
    description="Returns product details based on ID",
    response_description="Product data",
    tags=["Products"]
)
async def get_product(
    product_id: int = Path(..., title="Product ID", ge=1),
    include_reviews: bool = Query(False, description="Include reviews")
):
    """
    Retrieve detailed product information:

    - **product_id**: unique product ID
    - **include_reviews**: optionally include user reviews
    """
    return {"product_id": product_id, "include_reviews": include_reviews}

Every endpoint, parameter, and model is automatically visible in the documentation without writing any extra code.

Pydantic Models - Data Validation and Serialization#

FastAPI uses the Pydantic library for input and output data validation. Pydantic models define the shape of data and validation rules:

from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime

class Address(BaseModel):
    street: str = Field(..., min_length=3, max_length=200)
    city: str = Field(..., min_length=2, max_length=100)
    zip_code: str = Field(..., pattern=r"^\d{5}(-\d{4})?$")
    country: str = "US"

class UserCreate(BaseModel):
    username: str = Field(
        ...,
        min_length=3,
        max_length=50,
        description="Unique username"
    )
    email: EmailStr
    password: str = Field(..., min_length=8)
    full_name: Optional[str] = None
    age: int = Field(..., ge=18, le=120)
    address: Optional[Address] = None
    tags: List[str] = []

    @validator("password")
    def password_strength(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain an uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain a digit")
        return v

    class Config:
        json_schema_extra = {
            "example": {
                "username": "john_doe",
                "email": "john@example.com",
                "password": "MyPassword123",
                "full_name": "John Doe",
                "age": 30,
                "tags": ["developer", "python"]
            }
        }

class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    full_name: Optional[str]
    created_at: datetime

    class Config:
        from_attributes = True

Using models in endpoints:

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.post(
    "/users/",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Users"]
)
async def create_user(user: UserCreate):
    # Pydantic automatically validates input data
    # Invalid data returns HTTP 422 with a description of the issues
    db_user = save_to_database(user)
    return db_user

@app.get("/users/{user_id}", response_model=UserResponse, tags=["Users"])
async def get_user(user_id: int):
    user = get_from_database(user_id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail=f"User with ID {user_id} not found"
        )
    return user

Path Parameters, Query Parameters, and Request Bodies#

FastAPI provides a flexible parameter system:

from fastapi import FastAPI, Query, Path, Body
from typing import Optional, List

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    # Path parameter - required
    item_id: int = Path(..., title="Item ID", ge=1),

    # Query parameters - optional
    q: Optional[str] = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0, description="Skip N results"),
    limit: int = Query(10, ge=1, le=100, description="Result limit"),

    # List values in query string: ?tags=python&tags=fastapi
    tags: List[str] = Query([], description="Filter by tags"),

    # Sorting
    sort_by: Optional[str] = Query(
        None,
        regex="^(name|price|date)$",
        description="Sort field"
    )
):
    return {
        "item_id": item_id,
        "q": q,
        "skip": skip,
        "limit": limit,
        "tags": tags,
        "sort_by": sort_by
    }

@app.put("/items/{item_id}")
async def update_item(
    item_id: int = Path(..., ge=1),
    item: ItemUpdate = Body(...),
    importance: int = Body(1, ge=1, le=5)
):
    return {"item_id": item_id, "item": item, "importance": importance}

Async/Await - Asynchronous Programming#

FastAPI natively supports asynchronous programming, allowing it to handle many requests concurrently without blocking:

import httpx
import asyncio
from fastapi import FastAPI

app = FastAPI()

# Async endpoint - does not block the server
@app.get("/async-data")
async def fetch_multiple_sources():
    async with httpx.AsyncClient() as client:
        # Parallel HTTP requests
        tasks = [
            client.get("https://api.service1.com/data"),
            client.get("https://api.service2.com/data"),
            client.get("https://api.service3.com/data"),
        ]
        responses = await asyncio.gather(*tasks)

    return {
        "service1": responses[0].json(),
        "service2": responses[1].json(),
        "service3": responses[2].json(),
    }

# Synchronous endpoint - runs in a thread pool
@app.get("/sync-data")
def fetch_sync_data():
    # Blocking operations (e.g., CPU-intensive)
    # FastAPI automatically runs this in a separate thread
    result = heavy_computation()
    return {"result": result}

Dependency Injection System#

FastAPI includes a built-in dependency injection system that allows for elegant management of shared resources and logic across endpoints:

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional

app = FastAPI()

# Simple dependency
async def common_parameters(
    q: Optional[str] = None,
    skip: int = 0,
    limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}

# Dependency with sub-dependency
async def get_db():
    db = SessionLocal()
    try:
        yield db  # Generator - automatic cleanup
    finally:
        db.close()

async def get_current_user(
    token: str = Header(..., alias="Authorization")
):
    user = decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def get_current_active_user(
    current_user = Depends(get_current_user)
):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# Using dependencies in an endpoint
@app.get("/items/")
async def read_items(
    commons: dict = Depends(common_parameters),
    db = Depends(get_db),
    user = Depends(get_current_active_user)
):
    items = db.query(Item).offset(commons["skip"]).limit(commons["limit"]).all()
    return items

# Dependencies at the router level
from fastapi import APIRouter

router = APIRouter(
    prefix="/admin",
    tags=["Admin"],
    dependencies=[Depends(get_current_active_user)]
)

@router.get("/stats")
async def admin_stats(db = Depends(get_db)):
    return {"total_users": db.query(User).count()}

Authentication with OAuth2 and JWT#

FastAPI has built-in support for OAuth2 with JWT tokens:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user(username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

Database Integration with SQLAlchemy#

FastAPI works seamlessly with SQLAlchemy, both in synchronous and asynchronous modes:

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import FastAPI, Depends, HTTPException
from datetime import datetime
from typing import List

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy Model
class UserDB(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)
    hashed_password = Column(String(200))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

Base.metadata.create_all(bind=engine)

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

# CRUD operations
@app.post("/users/", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Check if user exists
    db_user = db.query(UserDB).filter(
        UserDB.email == user.email
    ).first()
    if db_user:
        raise HTTPException(
            status_code=400,
            detail="Email already registered"
        )

    # Create user
    hashed_password = pwd_context.hash(user.password)
    db_user = UserDB(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/", response_model=List[UserResponse])
def list_users(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    users = db.query(UserDB).offset(skip).limit(limit).all()
    return users

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(user)
    db.commit()

Background Tasks#

FastAPI allows you to run background tasks without blocking the HTTP response:

from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import JSONResponse
import smtplib
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

def send_email(email: str, subject: str, body: str):
    """Send email - background task"""
    logger.info(f"Sending email to {email}")
    with smtplib.SMTP("smtp.example.com") as server:
        server.sendmail("noreply@example.com", email,
                       f"Subject: {subject}\n\n{body}")
    logger.info(f"Email sent to {email}")

def generate_report(report_id: int, params: dict):
    """Generate report - time-consuming operation"""
    logger.info(f"Generating report {report_id}")
    result = process_data(params)
    save_report(report_id, result)
    logger.info(f"Report {report_id} ready")

@app.post("/users/register")
async def register_user(
    user: UserCreate,
    background_tasks: BackgroundTasks
):
    # Create user immediately
    db_user = create_user_in_db(user)

    # Send welcome email in the background
    background_tasks.add_task(
        send_email,
        email=user.email,
        subject="Welcome!",
        body="Thank you for registering with our service."
    )

    return {"id": db_user.id, "message": "Registered successfully"}

@app.post("/reports/")
async def create_report(
    params: ReportParams,
    background_tasks: BackgroundTasks
):
    report_id = create_report_entry()
    background_tasks.add_task(generate_report, report_id, params.dict())
    return {"report_id": report_id, "status": "processing"}

WebSocket - Real-Time Communication#

FastAPI offers built-in WebSocket support for bidirectional communication:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

class ConnectionManager:
    """WebSocket connection manager"""

    def __init__(self):
        self.active_connections: dict[str, List[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.active_connections:
            self.active_connections[room] = []
        self.active_connections[room].append(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        self.active_connections[room].remove(websocket)

    async def broadcast(self, message: str, room: str):
        for connection in self.active_connections.get(room, []):
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat/{room}")
async def websocket_chat(websocket: WebSocket, room: str):
    await manager.connect(websocket, room)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            # Broadcast to everyone in the room
            await manager.broadcast(
                json.dumps({
                    "user": message["user"],
                    "text": message["text"],
                    "room": room
                }),
                room
            )
    except WebSocketDisconnect:
        manager.disconnect(websocket, room)
        await manager.broadcast(
            json.dumps({"system": "A user has left the chat"}),
            room
        )

@app.websocket("/ws/notifications/{user_id}")
async def websocket_notifications(websocket: WebSocket, user_id: int):
    await websocket.accept()
    try:
        while True:
            # Send real-time notifications
            notification = await get_next_notification(user_id)
            await websocket.send_json(notification)
    except WebSocketDisconnect:
        pass

Testing with TestClient#

FastAPI provides a TestClient based on httpx for easy API testing:

from fastapi.testclient import TestClient
from main import app
import pytest

client = TestClient(app)

class TestUserEndpoints:

    def test_create_user(self):
        response = client.post(
            "/users/",
            json={
                "username": "testuser",
                "email": "test@example.com",
                "password": "TestPassword123",
                "age": 25
            }
        )
        assert response.status_code == 201
        data = response.json()
        assert data["username"] == "testuser"
        assert data["email"] == "test@example.com"
        assert "id" in data
        assert "password" not in data  # Password should not be in the response

    def test_create_user_invalid_email(self):
        response = client.post(
            "/users/",
            json={
                "username": "testuser",
                "email": "not-an-email",
                "password": "TestPassword123",
                "age": 25
            }
        )
        assert response.status_code == 422  # Validation error

    def test_get_user_not_found(self):
        response = client.get("/users/99999")
        assert response.status_code == 404

    def test_list_users_pagination(self):
        response = client.get("/users/?skip=0&limit=10")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)
        assert len(data) <= 10

# WebSocket test
def test_websocket_chat():
    with client.websocket_connect("/ws/chat/test-room") as websocket:
        websocket.send_json({"user": "tester", "text": "Hello!"})
        data = websocket.receive_json()
        assert data["user"] == "tester"
        assert data["text"] == "Hello!"

# Test with mocked dependencies
from unittest.mock import MagicMock

def test_with_mock_db():
    mock_db = MagicMock()
    mock_db.query.return_value.all.return_value = []

    app.dependency_overrides[get_db] = lambda: mock_db

    response = client.get("/users/")
    assert response.status_code == 200

    # Clear override
    app.dependency_overrides.clear()

# Test with authentication
def test_protected_endpoint():
    # Without token
    response = client.get("/users/me")
    assert response.status_code == 401

    # With valid token
    token = create_access_token(data={"sub": "testuser"})
    response = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200

Performance Comparison#

FastAPI stands out among other popular frameworks:

| Framework | Language | Req/s (JSON) | Req/s (DB query) | Async | |-----------|----------|-------------|-------------------|-------| | FastAPI | Python | ~32,000 | ~12,000 | Yes | | Flask | Python | ~8,000 | ~4,000 | No* | | Django REST | Python | ~5,000 | ~3,500 | Partial | | Express.js | Node.js | ~35,000 | ~14,000 | Yes | | Gin | Go | ~85,000 | ~35,000 | Yes |

*Flask with gevent can handle async, but not natively.

FastAPI is 3-4x faster than Flask and 5-6x faster than Django REST Framework in standard benchmarks. Compared to Express.js, it achieves comparable performance, which is an impressive result for an interpreted language.

Key advantages of FastAPI:

  • Vs Flask - automatic validation, documentation, async, type safety
  • Vs Django REST - significantly higher performance, simpler setup, better async
  • Vs Express.js - better validation, automatic documentation, type hints

Deployment with Docker and Uvicorn#

Professional FastAPI deployment requires proper ASGI server configuration and containerization:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY ./app ./app

# Create non-root user
RUN adduser --disabled-password --no-create-home appuser
USER appuser

# Run with Uvicorn
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Docker Compose configuration with database and cache:

version: "3.8"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  db:
    image: postgres:16-alpine
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=mydb
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

For production environments, using Gunicorn as a process manager is recommended:

gunicorn app.main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --access-logfile - \
  --error-logfile - \
  --timeout 120

Production Project Structure#

A well-organized FastAPI project should follow this structure:

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI application
│   ├── config.py             # Configuration (env variables)
│   ├── database.py           # Database connection
│   ├── dependencies.py       # Shared dependencies
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── users.py
│   │   ├── products.py
│   │   └── auth.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py           # SQLAlchemy models
│   │   └── product.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py           # Pydantic models
│   │   └── product.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py   # Business logic
│   │   └── email_service.py
│   └── middleware/
│       ├── __init__.py
│       └── logging.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_users.py
│   └── test_products.py
├── alembic/                   # Database migrations
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env

Summary#

FastAPI is an exceptionally mature framework that combines high performance with an outstanding developer experience. With automatic validation, documentation, and typing, the API development process is faster and less error-prone. Native async/await support, the dependency injection system, WebSocket handling, and seamless database integration make it an ideal choice for both simple projects and complex enterprise systems.

Key reasons to choose FastAPI:

  • Development speed - less code, automatic documentation
  • Performance - comparable to Node.js and Go
  • Type safety - validation at every level
  • Ecosystem - integration with Pydantic, SQLAlchemy, Celery, and many more
  • Production readiness - easy deployment with Docker and Uvicorn

Need a Professional API?#

At MDS Software Solutions Group, we specialize in:

  • Designing and building APIs with FastAPI and Python
  • Microservices architecture and distributed systems
  • Database integration and third-party system connectivity
  • Production deployments with Docker and Kubernetes
  • Performance optimization and API security audits

Contact us to discuss your project!

Author
MDS Software Solutions Group

Team of programming experts specializing in modern web technologies.

FastAPI - Building Modern APIs in Python. A Complete Guide | MDS Software Solutions Group | MDS Software Solutions Group