Learnixo
Back to blog
Backend Systemsbeginner

Python & FastAPI: Build Production REST APIs

Build production-ready REST APIs with Python and FastAPI. Covers Python fundamentals, async Python, Pydantic models, dependency injection, JWT auth, database integration with SQLAlchemy, and deployment.

LearnixoApril 13, 20269 min read
View Source
PythonFastAPIREST APIPydanticSQLAlchemyasync
Share:𝕏

Why FastAPI?

FastAPI is the fastest-growing Python web framework. It's used at Microsoft, Uber, Netflix, and hundreds of startups. What makes it special:

  • Performance: on par with Node.js and Go — built on async Python (ASGI)
  • Auto documentation: Swagger UI and ReDoc generated from your code
  • Type safety: built-in validation via Pydantic — catch bugs before they ship
  • Developer experience: auto-completion, inline validation errors, zero boilerplate

If you know Python, you can build a production API in under an hour with FastAPI.

Python Essentials for FastAPI

Before FastAPI, you need to understand these Python concepts.

Type hints

Python
# Python 3.10+
def greet(name: str, age: int) -> str:
    return f"Hello {name}, you are {age}"

from typing import Optional, List, Dict, Union

def get_user(user_id: int, include_orders: bool = False) -> Optional[dict]:
    ...

# Modern union syntax (Python 3.10+)
def process(value: int | str | None) -> str:
    ...

Dataclasses and Pydantic models

Python
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    id: int
    name: str
    email: str
    created_at: datetime = field(default_factory=datetime.utcnow)

List/dict comprehensions

Python
# List comprehension
squares = [x**2 for x in range(10)]
evens = [n for n in numbers if n % 2 == 0]

# Dict comprehension
email_map = {user.id: user.email for user in users}

# Generator (lazy, memory efficient)
total = sum(item.price * item.qty for item in cart_items)

async / await

Python
import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # simulates I/O
    return {"id": user_id, "name": "Alice"}

async def main():
    # Run concurrently
    users = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )

asyncio.run(main())

Decorators

Python
import functools
import time

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} took {time.time() - start:.2f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

Project Setup

Bash
# Create virtual environment
python -m venv venv
source venv/bin/activate        # Linux/Mac
venv\Scripts\activate           # Windows

# Install dependencies
pip install fastapi uvicorn[standard] sqlalchemy alembic pydantic-settings
pip install python-jose[cryptography] passlib[bcrypt] httpx

# Create project structure
mkdir -p app/{models,schemas,routers,services,core}
touch app/{__init__,main,database}.py
orderflow/
├── app/
│   ├── main.py           # FastAPI app, startup
│   ├── database.py       # SQLAlchemy engine
│   ├── models/           # SQLAlchemy ORM models
│   ├── schemas/          # Pydantic request/response models
│   ├── routers/          # Route handlers
│   ├── services/         # Business logic
│   └── core/
│       ├── config.py     # Settings from environment
│       ├── security.py   # JWT, hashing
│       └── deps.py       # Dependency injection
├── alembic/              # Database migrations
├── tests/
└── requirements.txt

FastAPI Application

Python
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import orders, auth, customers
from app.database import create_tables

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await create_tables()
    yield
    # Shutdown (cleanup)

app = FastAPI(
    title="OrderFlow API",
    description="Order management REST API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://learnixo.io", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router,      prefix="/api/v1/auth",      tags=["auth"])
app.include_router(customers.router, prefix="/api/v1/customers",  tags=["customers"])
app.include_router(orders.router,    prefix="/api/v1/orders",     tags=["orders"])

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

Pydantic Models (Schemas)

Pydantic validates and serializes data automatically.

Python
# app/schemas/order.py
from pydantic import BaseModel, field_validator, model_validator
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import List

class OrderStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

class OrderItemCreate(BaseModel):
    product_id: int
    quantity: int

    @field_validator("quantity")
    @classmethod
    def quantity_must_be_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

class OrderCreate(BaseModel):
    customer_id: int
    items: List[OrderItemCreate]
    notes: str | None = None

    @model_validator(mode="after")
    def items_not_empty(self) -> "OrderCreate":
        if not self.items:
            raise ValueError("Order must have at least one item")
        return self

class OrderItemResponse(BaseModel):
    id: int
    product_id: int
    product_name: str
    quantity: int
    unit_price: Decimal

    model_config = {"from_attributes": True}  # allows ORM model → schema

class OrderResponse(BaseModel):
    id: int
    customer_id: int
    customer_name: str
    status: OrderStatus
    total: Decimal
    items: List[OrderItemResponse]
    created_at: datetime

    model_config = {"from_attributes": True}

class PagedResponse(BaseModel):
    items: List[OrderResponse]
    page: int
    page_size: int
    total: int
    total_pages: int

SQLAlchemy Models

Python
# app/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings

engine = create_async_engine(settings.database_url, echo=settings.debug)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncSession:  # dependency
    async with AsyncSessionLocal() as session:
        yield session

# app/models/order.py
from sqlalchemy import String, Integer, Numeric, ForeignKey, DateTime, Enum as SAEnum
from sqlalchemy.orm import relationship, Mapped, mapped_column
from datetime import datetime
from app.database import Base

class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    tier: Mapped[str] = mapped_column(String(20), default="free")
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    is_active: Mapped[bool] = mapped_column(default=True)

    orders: Mapped[list["Order"]] = relationship("Order", back_populates="customer")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    status: Mapped[str] = mapped_column(String(20), default="pending")
    total: Mapped[float] = mapped_column(Numeric(10, 2), default=0)
    notes: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    customer: Mapped["Customer"] = relationship("Customer", back_populates="orders")
    items: Mapped[list["OrderItem"]] = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")

Router with Dependency Injection

Python
# app/routers/orders.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.order import OrderCreate, OrderResponse, PagedResponse
from app.services.order_service import OrderService
from app.core.deps import get_current_user, require_role
from app.models.user import User

router = APIRouter()

@router.get("/", response_model=PagedResponse)
async def list_orders(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    return await service.get_paged(page, page_size, status, current_user.id)

@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(
    order_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    order = await service.get_by_id(order_id, current_user.id)
    if not order:
        raise HTTPException(status_code=404, detail=f"Order {order_id} not found")
    return order

@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
async def create_order(
    payload: OrderCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    return await service.create(payload, current_user.id)

@router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_order(
    order_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    service = OrderService(db)
    deleted = await service.delete(order_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Order not found")

JWT Authentication

Python
# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=60))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.jwt_secret, algorithm="HS256")

def decode_token(token: str) -> dict:
    return jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])

# app/core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.core.security import decode_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await get_user_by_id(db, user_id)
    if user is None:
        raise credentials_exception
    return user

def require_role(role: str):
    async def checker(user = Depends(get_current_user)):
        if user.role != role:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return user
    return checker

# app/routers/auth.py
@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
):
    user = await authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect email or password")

    token = create_access_token({"sub": str(user.id), "role": user.role})
    return {"access_token": token, "token_type": "bearer"}

Settings and Configuration

Python
# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # Database
    database_url: str = "postgresql+asyncpg://user:pass@localhost/orderflow"

    # JWT
    jwt_secret: str = "change-me-in-production"
    jwt_algorithm: str = "HS256"
    jwt_expire_minutes: int = 60

    # App
    debug: bool = False
    environment: str = "development"

    # Email
    smtp_host: str = "smtp.gmail.com"
    smtp_port: int = 587
    smtp_user: str = ""
    smtp_password: str = ""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()
Bash
# .env (never commit this)
DATABASE_URL=postgresql+asyncpg://postgres:password@localhost/orderflow
JWT_SECRET=super-secret-key-minimum-32-characters
ENVIRONMENT=production
DEBUG=false

Global Error Handling

Python
# app/main.py
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "errors": exc.errors(),
        },
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    import logging
    logging.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

Background Tasks

Python
from fastapi import BackgroundTasks

@router.post("/orders", status_code=201)
async def create_order(
    payload: OrderCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    order = await OrderService(db).create(payload, current_user.id)

    # Fire and forget — doesn't delay the response
    background_tasks.add_task(send_confirmation_email, order.id, current_user.email)

    return order

async def send_confirmation_email(order_id: int, email: str):
    # runs after the response is sent
    await email_service.send(email, f"Order #{order_id} confirmed")

Database Migrations with Alembic

Bash
# Initialize Alembic
alembic init alembic

# Create migration
alembic revision --autogenerate -m "create_orders_table"

# Run migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1
Python
# alembic/env.py
from app.database import Base
from app.models import *  # import all models so Alembic finds them

target_metadata = Base.metadata

Testing

Python
# tests/test_orders.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

@pytest.fixture
async def auth_headers(client):
    response = await client.post("/api/v1/auth/login", data={
        "username": "test@example.com",
        "password": "testpassword",
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

@pytest.mark.asyncio
async def test_create_order(client, auth_headers):
    response = await client.post(
        "/api/v1/orders",
        json={"customer_id": 1, "items": [{"product_id": 1, "quantity": 2}]},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["status"] == "pending"

@pytest.mark.asyncio
async def test_get_order_not_found(client, auth_headers):
    response = await client.get("/api/v1/orders/99999", headers=auth_headers)
    assert response.status_code == 404

Running in Production

DOCKERFILE
# Dockerfile
FROM python:3.12-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", "8000", "--workers", "4"]
Bash
# Run locally
uvicorn app.main:app --reload --port 8000

# View auto-generated docs
open http://localhost:8000/docs      # Swagger UI
open http://localhost:8000/redoc     # ReDoc

What to Learn Next

  • Python Interview Questions: 100 Q&As from beginner to senior
  • Docker & Kubernetes: containerise and deploy this API
  • AI Systems Engineering: add LLM capabilities to your FastAPI service

Enjoyed this article?

Explore the Backend Systems learning path for more.

Subscribe for weekly tutorialsFull source code

Found this helpful?

Share:𝕏

Leave a comment

Have a question, correction, or just found this helpful? Leave a note below.