Building a REST API with FastAPI

FastAPI is a modern, fast web framework for building APIs with Python. It's designed to be easy to use, fast to code, and ready for production. Let's explore how to build a robust REST API using FastAPI.

Getting Started with FastAPI 🚀

First, let's set up our development environment and create a basic API:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

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

@app.post("/items/")
async def create_item(item: Item):
    return item

Project Structure 📁

Let's organize our API with a clean, maintainable structure:

# app/main.py
from fastapi import FastAPI
from app.routers import items, users
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

Configuration Management

# app/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "FastAPI Example"
    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str

    class Config:
        env_file = ".env"

settings = Settings()

Database Integration 🗄️

Let's implement database models and connections using SQLAlchemy:

# app/db/base.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(SQLALCHEMY_DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)
Base = declarative_base()

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Database Models

# app/models/item.py
from sqlalchemy import Column, Integer, String, Float
from app.db.base import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    price = Column(Float)
    tax = Column(Float, nullable=True)

API Routes Implementation

Items Router

# app/routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from app.db.base import get_db
from app.schemas.item import ItemCreate, ItemResponse
from app.crud import items

router = APIRouter()

@router.post("/", response_model=ItemResponse)
async def create_item(
    item: ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    return await items.create(db, item)

@router.get("/", response_model=List[ItemResponse])
async def read_items(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db)
):
    return await items.get_multi(db, skip=skip, limit=limit)

@router.get("/{item_id}", response_model=ItemResponse)
async def read_item(
    item_id: int,
    db: AsyncSession = Depends(get_db)
):
    item = await items.get(db, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Authentication and Authorization 🔒

Implement JWT-based authentication:

# app/core/auth.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException, status

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

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

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
    return username

Request Validation and Error Handling

# app/schemas/item.py
from pydantic import BaseModel, Field, validator

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    description: str | None = Field(None, max_length=500)
    price: float = Field(..., gt=0)
    tax: float | None = Field(None, ge=0)

    @validator('price')
    def validate_price(cls, v):
        if v <= 0:
            raise ValueError('Price must be greater than zero')
        return round(v, 2)

class ItemCreate(ItemBase):
    pass

class ItemResponse(ItemBase):
    id: int

    class Config:
        from_attributes = True

Middleware Implementation

# app/middleware/logging.py
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

Dependency Injection

# app/dependencies.py
from fastapi import Header, HTTPException
from typing import Annotated

async def get_token_header(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

Testing 🧪

# tests/test_items.py
from fastapi.testclient import TestClient
from app.main import app
import pytest

client = TestClient(app)

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "price": 45.50}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test Item"
    assert data["price"] == 45.50

@pytest.mark.asyncio
async def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    data = response.json()
    assert "name" in data

API Documentation 📚

FastAPI automatically generates OpenAPI (Swagger) documentation. Enhance it with detailed descriptions:

@app.get(
    "/items/{item_id}",
    response_model=ItemResponse,
    responses={
        404: {"description": "Item not found"},
        200: {
            "description": "Successful Response",
            "content": {
                "application/json": {
                    "example": {
                        "id": 1,
                        "name": "Example Item",
                        "price": 45.50
                    }
                }
            }
        }
    }
)
async def read_item(item_id: int):
    """
    Retrieve an item by its ID.

    - **item_id**: The ID of the item to retrieve
    """
    pass

Performance Optimization ⚡

Caching Implementation

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

@router.get("/cached-items/")
@cache(expire=60)
async def get_cached_items():
    # Expensive operation here
    return await get_items_from_db()

Background Tasks

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("app.log", mode="a") as log:
        log.write(f"{message}\n")

@app.post("/items/")
async def create_item(
    item: Item,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, f"Created item {item.name}")
    return {"message": "Item created"}

Deployment Configuration 🚀

Docker Setup

FROM python:3.11-slim

WORKDIR /app

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

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

Production Settings

# app/core/config.py
class ProductionSettings(Settings):
    class Config:
        env_file = ".env.prod"

    POSTGRES_SERVER: str
    REDIS_URL: str
    SENTRY_DSN: str | None = None

Best Practices 📝

1. Use Async Where Appropriate

@router.get("/items/")
async def read_items():
    # Use async for I/O-bound operations
    return await database.fetch_all(query)

2. Implement Rate Limiting

from fastapi import Request
from fastapi.responses import JSONResponse
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.get("/limited")
@limiter.limit("5/minute")
async def limited_route(request: Request):
    return {"message": "Success"}

3. Structured Logging

import structlog

logger = structlog.get_logger()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(
        "request_started",
        path=request.url.path,
        method=request.method
    )
    response = await call_next(request)
    return response

Conclusion

FastAPI provides a robust framework for building modern APIs. Remember to:

• Structure your project properly
• Implement proper authentication and authorization
• Use type hints and Pydantic models
• Write comprehensive tests
• Optimize performance where needed
• Follow REST best practices
• Document your API thoroughly

These practices will help you build maintainable, scalable, and efficient APIs with FastAPI.