FastAPI Complete Guide

Build Production-Ready APIs with FastAPI

Learn how to structure a FastAPI application with proper layered architecture, async operations, configuration management, logging, and production best practices.

FastAPI Architecture

Introduction

If you've been building APIs with Python, you've probably experienced the pain of writing endless boilerplate code, manually validating data, and maintaining documentation that's always out of sync with your actual code.

That's why FastAPI is the go-to framework for building APIs. In this article, I'll walk you through building a complete Order Management System, focusing not just on what to build, but why we build it in a certain way.

By the end, you'll understand how to structure production grade API’s.

Why FastAPI:

Before FastAPI , building a simple API endpoint meant writing code like this:

@app.route('/orders', methods=['POST'])
def create_order():
    data = request.get_json()

    # Manual validation (20+ lines)
    if not data or 'product_name' not in data:
        return {"error": "Product name required"}, 400
    if data.get('quantity', 0) <= 0:
        return {"error": "Quantity must be positive"}, 400

    order = Order(user_id=current_user.id, **data)
    db.session.add(order)
    db.session.commit()

    return jsonify({
        "id": order.id,
        "product_name": order.product_name,
    }), 201

FastAPI reduces this to

@router.post(
    "/",
    response_model=OrderResponse,
    status_code=201
)
async def create_order(
    order: OrderCreate,
    current_user: User = Depends(get_current_user),
    service: OrderService = Depends(get_order_service),
):
    return await service.create_order(
        current_user.id, order
    )

FastAPI aims to provide

• ⚡ Fast to code: Build features 2–3x faster compared to other Python frameworks
• 🛡 Auto validation: Pydantic rejects bad requests before they reach your code
• 📘 Auto docs: Swagger UI generated automatically as you build
• 🚀 Async ready: Built in from the start — no workarounds needed
• 🧠 Fewer bugs: Type hints catch errors early
• 🎯 Easy to learn: Great editor support, intuitive design
• 🔧 Production ready: Fully compatible with OpenAPI and JSON Schema

Architecture

FastAPI Architecture

Project Structure

Here’s how we organize our FastAPI application:

fastapi/                                                                                                                                                                    
  ├── Dockerfile                                                                                                                                                                     
  ├── docker-compose.yml                                                                                                                                                             
  ├── requirements.txt                                                                                                                                                               
  ├── pytest.ini                                                                                                                                                                                                                                                                                                                                      
  │                                                                                                                                                                                  
  ├── config/                                                                                                                                                                        
  │   └── logging.ini                                                                                                                                                                
  │
  ├── var/
  │   └── logs/
  │       ├── cog.log
  │       └── cog-error.log
  │
  ├── tests/
  │   ├── conftest.py
  │   ├── test_order_api.py         
  │   └── test_health.py
  │
  └── app/
      ├── main.py
      ├── api/
      │   ├── order_api.py           # /api/orders/ endpoints
      │   └── v1/
      │       ├── router.py
      │       └── endpoints/
      │           ├── auth.py        # /login, /register
      │           └── users.py       # /current-logged-user
      ├── core/
      │   ├── settings.py            # pydantic-settings config
      │   ├── security.py            # JWT + bcrypt
      │   └── deps.py                # get_current_user dependency
      ├── db/
      │   └── database.py            # async engine, get_db, Base
      ├── middleware/
      │   └── logging_middleware.py
      ├── models/
      │   ├── __init__.py            # registers models with Base.metadata
      │   ├── order.py               # FK → users
      │   └── user.py
      ├── schemas/
      │   ├── order_schema.py        # OrderCreate, OrderUpdate, OrderResponse
      │   └── user.py
      ├── services/
      │   └── order_service.py
      └── utils/
          ├── constants.py
          └── exceptions.py          # OrderNotFoundException, handlers

Building the Foundation

1. Constants & Paths
   Centralise all file paths in one place. This means changing a path requires editing a single file — not checking across the entire code.

#app/utils/constants.py
import os

PROJECT_ROOT        = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
LOGS_DIR            = os.path.join(PROJECT_ROOT, "var", "logs")
CONFIG_DIR          = os.path.join(PROJECT_ROOT, "config")
LOGGING_CONFIG_FILE = os.path.join(CONFIG_DIR, "logging.ini")

2.Logging Configuration

We set up logging before the database and models — so every layer of the application can use logger from the start. All config lives in config/logging.ini. No inline basicConfig() calls anywhere in the code.

#config/logging.ini
[loggers]
keys=root, fastapi.app

[handlers]
keys=debugHandler, errorHandler, consoleHandler

[formatters]
keys=simpleFormatter, appFormatter

[logger_fastapi.app]
level=DEBUG
handlers=debugHandler, errorHandler, consoleHandler
qualname=fastapi.app
propagate=0

[handler_debugHandler]
class=handlers.RotatingFileHandler
level=INFO
formatter=appFormatter
args=('var/logs/cog.log', 'a', 5242880, 5, 'utf-8')

[handler_errorHandler]
class=handlers.RotatingFileHandler
level=ERROR
formatter=appFormatter
args=('var/logs/cog-error.log', 'a', 5242880, 5, 'utf-8')

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[formatter_simpleFormatter]
format = %(asctime)s | %(name)s | %(levelname)-8s | %(message)s
datefmt=%Y-%m-%d %H:%M:%S
class=logging.Formatter

[formatter_appFormatter]
format = [%(asctime)s] | %(levelname)s | %(module)s | %(funcName)s | %(message)s
datefmt=%Y-%m-%d %H:%M:%S
class=logging.Formatter

In any module, add a logger with one line — no config needed because logging.ini handles everything:

import logging
logger = logging.getLogger("fastapi.app")
logger.debug("Detailed trace for debugging")
logger.info("Creating order for user_id=%s", user_id)
logger.warning("Order id=%s not found", order_id)
logger.error("Unexpected error: %s", str(e))

We also add a middleware that automatically logs every incoming request and outgoing response:

#app/middleware/logging_middleware.py
import logging
from starlette.middleware.base import BaseHTTPMiddleware
logger = logging.getLogger("fastapi.app")
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        logger.info("Request: %s %s", request.method, request.url.path)
        response = await call_next(request)
        logger.info("Response: %s", response.status_code)
        return response

3. Async Database Setup
SQLAlchemy’s async engine with asyncpg ensures database calls never block the event loop. We use PostgreSQL in production and SQLite in tests.

#app/db/database.py
import logging
import os
from typing import AsyncGenerator

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base

logger = logging.getLogger("fastapi.app")

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres:postgres@localhost:5432/orders")

engine            = create_async_engine(DATABASE_URL, echo=False, future=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base              = declarative_base()


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()


async def create_tables_if_not_exist():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

get_db() leverages FastAPI's dependency injection to provide a fresh session per request and close it automatically — even if an exception is raised.

4. Defining the Data Models
We have two models: User (who logs in) and Order (what they place). The critical field is user_id — a foreign key that ties every order to the user who created it.

#app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean
from app.db.database import Base

class User(Base):
    __tablename__ = "users"

    id              = Column(Integer, primary_key=True, autoincrement=True)
    email           = Column(String(100), unique=True, nullable=False, index=True)
    hashed_password = Column(String, nullable=False)
    is_active       = Column(Boolean, default=True, nullable=False)

#app/models/order.py
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.sql import func
from app.db.database import Base

class Order(Base):
    __tablename__ = "orders"

    id           = Column(Integer, primary_key=True, index=True, autoincrement=True)
    user_id      = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    product_name = Column(String(200), nullable=False)
    quantity     = Column(Integer, nullable=False, default=1)
    unit_price   = Column(Float, nullable=False)
    status       = Column(String(20), nullable=False, default="pending")
    created_at   = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)

Note : user_id is the key design decision. Every order has an owner. Without this, you’d have no way to filter orders per user — auth would be decorative rather than functional.

5. Request/Response Schemas with Pydantic
We use an OrderStatus enum so status values are validated automatically — no typos, no invalid states.

#app/schemas/order_schema.py
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from typing import Optional
from enum import Enum


class OrderStatus(str, Enum):
    pending   = "pending"
    confirmed = "confirmed"
    shipped   = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"


class OrderCreate(BaseModel):
    product_name: str   = Field(..., min_length=1, max_length=200)
    quantity:     int   = Field(..., gt=0)
    unit_price:   float = Field(..., gt=0)


class OrderUpdate(BaseModel):
    product_name: Optional[str]         = Field(None, min_length=1, max_length=200)
    quantity:     Optional[int]         = Field(None, gt=0)
    unit_price:   Optional[float]       = Field(None, gt=0)
    status:       Optional[OrderStatus] = None


class OrderResponse(BaseModel):
    id:           int
    user_id:      int
    product_name: str
    quantity:     int
    unit_price:   float
    status:       str
    created_at:   datetime

    model_config = ConfigDict(from_attributes=True)

• OrderCreate only asks for product details — user_id is never taken from the request body (it comes from the JWT token).
• OrderUpdate makes every field optional, enabling partial updates.
• OrderStatus enum prevents invalid status values at the type level.

6. The Service Layer: Where Business Logic Lives
Every method takes user_id as a parameter. This ensures data isolation — a user can only ever see, update, or delete their own orders. When something goes wrong the service raises a typed exception — it never returns None or False for error states.

#app/services/order_service.py
import logging
from typing import List

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.order import Order
from app.schemas.order_schema import OrderCreate, OrderUpdate
from app.utils.exceptions import OrderNotFoundException, OrderAlreadyExistsException

logger = logging.getLogger("fastapi.app")


class OrderService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create_order(self, user_id: int, order_data: OrderCreate) -> Order:
        logger.debug("Creating order for user_id=%s product=%s", user_id, order_data.product_name)

        # Prevent duplicate pending orders for the same product
        existing = await self.db.execute(
            select(Order).where(
                Order.user_id == user_id,
                Order.product_name == order_data.product_name,
                Order.status == "pending",
            )
        )
        if existing.scalar_one_or_none():
            raise OrderAlreadyExistsException(order_data.product_name)

        order = Order(user_id=user_id, **order_data.model_dump())
        self.db.add(order)
        await self.db.commit()
        await self.db.refresh(order)
        logger.debug("Order created with id=%s", order.id)
        return order

    async def get_orders(self, user_id: int) -> List[Order]:
        result = await self.db.execute(
            select(Order).where(Order.user_id == user_id).order_by(Order.created_at.desc())
        )
        return list(result.scalars().all())

    async def get_order(self, order_id: int, user_id: int) -> Order:
        result = await self.db.execute(
            select(Order).where(Order.id == order_id, Order.user_id == user_id)
        )
        order = result.scalar_one_or_none()
        if not order:
            raise OrderNotFoundException(order_id)
        return order

    async def update_order(self, order_id: int, user_id: int, order_data: OrderUpdate) -> Order:
        order = await self.get_order(order_id, user_id)  # raises OrderNotFoundException if missing
        for field, value in order_data.model_dump(exclude_unset=True).items():
            setattr(order, field, value)
        await self.db.commit()
        await self.db.refresh(order)
        return order

    async def delete_order(self, order_id: int, user_id: int) -> None:
        order = await self.get_order(order_id, user_id)  # raises OrderNotFoundException if missing
        await self.db.delete(order)
        await self.db.commit()

The key pattern: get_order queries by both order_id AND user_id. If User A tries to access User B's order, scalar_one_or_none() returns None and an OrderNotFoundException is raised — just as if the order doesn't exist. No extra permission checks needed.

Why raise exceptions instead of returning None? Returning None forces every caller (the API layer) to check the result and manually raise HTTPException. A typed exception is raised once in the service and handled once by a registered handler — the API layer stays completely clean.

7. Exception Handling

Instead of scattering HTTPException raises throughout controllers, we define typed exception classes in one place and register FastAPI handlers that convert them to proper HTTP responses automatically.

#app/utils/exceptions.py
from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse


class OrderNotFoundException(Exception):
    """Raised when an order is not found for the given user."""

    def __init__(self, order_id: int):
        self.order_id = order_id
        self.message = f"Order with id {order_id} not found"
        super().__init__(self.message)


class OrderAlreadyExistsException(Exception):
    """Raised when the user already has a pending order for the same product."""

    def __init__(self, product_name: str):
        self.product_name = product_name
        self.message = f"You already have a pending order for '{product_name}'"
        super().__init__(self.message)


# ── Exception handlers ────────────────────────────────────────────────────────

async def order_not_found_exception_handler(request: Request, exc: OrderNotFoundException) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"error": "Not Found", "message": exc.message,
                 "order_id": exc.order_id, "path": str(request.url)},
    )


async def order_already_exists_exception_handler(request: Request, exc: OrderAlreadyExistsException) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={"error": "Conflict", "message": exc.message,
                 "product_name": exc.product_name, "path": str(request.url)},
    )


async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"error": "Validation Error", "message": "Invalid input data",
                 "details": exc.errors(), "path": str(request.url)},
    )


async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"error": "Internal Server Error", "message": "An unexpected error occurred",
                 "details": str(exc), "path": str(request.url)},
    )

The handlers are registered on the FastAPI app in main.py. The entire flow is:

Authentication & Security

In a typical “Hello World” example,there isnt any authentication.
Authentication is important without knowing who is logged in, we cannot know which orders to return.

When a user calls GET /api/orders/, the API reads their identity from the JWT token and filters the database by user_id. User A never sees User B orders.

Install the required packages:

Dependencies for Auth

pip install pydantic-settings python-jose[cryptography] bcrypt python-multipart

Use bcrypt directly — skip passlib. passlib is unmaintained and has a compatibility break with bcrypt 4.1+. Calling bcrypt.hashpw / bcrypt.checkpw directly works with any bcrypt version and has no extra dependencies.

The full auth flow:

Auth Flow

Settings

pydantic-settings reads from environment variables and .env files automatically. Every config value are passed here through os.getenv() calls.

#app/core/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
    APP_NAME: str = "Order Management System"
    ENV:      str = "local"
    DEBUG:    bool = True
    SECRET_KEY:                  str = ""
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    JWT_ALG:                     str = "HS256"
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/orders"
    CORS_ORIGINS: list[str] = ["*"]
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)
settings = Settings()

Why not os.getenv()? BaseSettings gives you type validation, default values, and IDE autocomplete. A missing required variable raises a clear error at startup instead of crashing at runtime.

Password Hashing & JWT

#app/core/security.py
import bcrypt
from datetime import datetime, timedelta, timezone
from jose import jwt
from app.core.settings import settings


def hash_password(plain: str) -> str:
    return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")

def verify_password(plain: str, hashed: str) -> bool:
    return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))

def create_access_token(sub: str, minutes: int | None = None) -> str:
    expire = datetime.now(timezone.utc) + timedelta(
        minutes=minutes or settings.ACCESS_TOKEN_EXPIRE_MINUTES
    )
    return jwt.encode(
        {"sub": sub, "exp": expire},
        settings.SECRET_KEY,
        algorithm=settings.JWT_ALG,
    )

• hash_password — encodes to bytes, hashes with bcrypt, stores as a UTF-8 string. Never stores the plain password.
• verify_password — re-encodes both strings to bytes and calls bcrypt.checkpw for a timing-safe comparison.
• create_access_token — signs a JWT embedding the user's email as sub.

Auth Dependency — the authentication to the controller

This is where we connect auth to the controller. It reads the JWT from the Authorization: Bearer ... header using HTTPBearer, decodes it, and returns the logged-in User object. Any route that uses Depends(get_current_user) gets the user's identity automatically — including their id for scoping orders.

HTTPBearer is used instead of OAuth2PasswordBearer so that the Swagger UI Authorize dialog simply shows a single token input field — no username, password, scopes, or client credentials.

#app/core/deps.py
import logging

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.settings import settings
from app.db.database import get_db

logger = logging.getLogger("fastapi.app")

bearer_scheme = HTTPBearer()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
    session: AsyncSession = Depends(get_db),
):
    from app.models.user import User  # local import avoids circular dependency

    token = credentials.credentials
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALG])
        email: str = payload.get("sub")
        if not email:
            raise JWTError()
    except JWTError:
        logger.warning("Invalid JWT token received")
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

    result = await session.execute(select(User).where(User.email == email))
    user = result.scalar_one_or_none()
    if not user:
        logger.warning("Token valid but user not found: %s", email)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

Auth Endpoints — register & login

Get Veenarao’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The login endpoint uses a custom LoginForm dependency (form data with only username and password).

The register endpoint validates that the password does not exceed 72 bytes — bcrypt’s limit — using a Pydantic field_validator, returning a clean 422 instead of a 500 crash.

#app/api/v1/endpoints/auth.py
import logging

from fastapi import APIRouter, Depends, Form, HTTPException, status
from pydantic import BaseModel, EmailStr, field_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import verify_password, create_access_token, hash_password
from app.db.database import get_db
from app.models.user import User

logger = logging.getLogger("fastapi.app")
router = APIRouter(tags=["Auth"])


class Token(BaseModel):
    access_token: str
    token_type:   str = "bearer"


class LoginForm:
    """Custom form — only exposes username and password in Swagger UI."""

    def __init__(
        self,
        username: str = Form(..., description="Your registered email address"),
        password: str = Form(...),
    ):
        self.username = username
        self.password = password


class RegisterRequest(BaseModel):
    email:    EmailStr
    password: str

    @field_validator("password")
    @classmethod
    def password_byte_length(cls, v: str) -> str:
        if len(v.encode("utf-8")) > 72:
            raise ValueError("Password must not exceed 72 characters")
        return v


@router.post("/login", response_model=Token, summary="Login")
async def login(
    form_data: LoginForm = Depends(),
    session: AsyncSession = Depends(get_db),
):
    result = await session.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()

    if not user:
        logger.warning("Login failed — user not found: %s", form_data.username)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

    try:
        password_ok = verify_password(form_data.password, user.hashed_password)
    except Exception as e:
        logger.error("Password verification error for %s: %s", form_data.username, e)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

    if not password_ok:
        logger.warning("Login failed — wrong password for: %s", form_data.username)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

    logger.info("User logged in: %s", user.email)
    return Token(access_token=create_access_token(sub=user.email))


@router.post("/register", status_code=status.HTTP_201_CREATED, summary="Register a new user")
async def register(
    data: RegisterRequest,
    session: AsyncSession = Depends(get_db),
):
    existing = await session.execute(select(User).where(User.email == data.email))
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")
    user = User(email=data.email, hashed_password=hash_password(data.password))
    session.add(user)
    await session.commit()
    logger.info("New user registered: %s", data.email)
    return {"id": user.id, "email": user.email}

API Versioning
Wire auth endpoints into a versioned router to maintain backward compatibility.

#app/api/v1/router.py
from fastapi import APIRouter
from .endpoints import auth, users
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(auth.router,  prefix="/auth")
api_router.include_router(users.router)

Why /api/v1/? Versioning lets you ship a v2 with breaking changes without removing v1 — existing clients keep working while you migrate them.

The Controller: Where Routing begins

Now that auth and exception handling are in place, the controller starts with routing the path. It uses Depends(get_current_user) to get the logged-in user, passes current_user.id to the service, and returns the result directly. There are no if not order checks and no manual HTTPException raises — the service raises a typed exception and the registered handler converts it to a JSON response.

#app/api/orders_api.PY
import logging

from fastapi import APIRouter, Depends, Path, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from app.core.deps import get_current_user
from app.db.database import get_db
from app.models.user import User
from app.schemas.order_schema import OrderCreate, OrderResponse, OrderUpdate
from app.services.order_service import OrderService

logger = logging.getLogger("fastapi.app")

router = APIRouter(
    prefix="/api/orders",
    tags=["Orders API"],
    dependencies=[Depends(get_current_user)],  # all routes require login
)


async def get_order_service(db: AsyncSession = Depends(get_db)) -> OrderService:
    return OrderService(db)


@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED,
             summary="Place a new order")
async def create_order(
    order: OrderCreate,
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user),
):
    logger.info("User %s placing order: product=%s", current_user.email, order.product_name)
    result = await service.create_order(current_user.id, order)
    logger.info("Order created with id=%s", result.id)
    return result


@router.get("/", response_model=List[OrderResponse], summary="List my orders")
async def get_my_orders(
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user),
):
    logger.info("Fetching orders for user %s", current_user.email)
    orders = await service.get_orders(current_user.id)
    logger.info("Found %d orders for user %s", len(orders), current_user.email)
    return orders


@router.get("/{order_id}", response_model=OrderResponse, summary="Get order by ID")
async def get_order(
    order_id: int = Path(..., gt=0, description="The ID of the order"),
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user),
):
    logger.info("Fetching order id=%s for user %s", order_id, current_user.email)
    return await service.get_order(order_id, current_user.id)


@router.put("/{order_id}", response_model=OrderResponse, summary="Update an order")
async def update_order(
    order_update: OrderUpdate,
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user),
    order_id: int = Path(..., gt=0, description="The ID of the order"),
):
    logger.info("Updating order id=%s for user %s", order_id, current_user.email)
    result = await service.update_order(order_id, current_user.id, order_update)
    logger.info("Order id=%s updated", order_id)
    return result


@router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT,
               summary="Cancel / delete an order")
async def delete_order(
    order_id: int = Path(..., gt=0, description="The ID of the order"),
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user),
):
    logger.info("Deleting order id=%s for user %s", order_id, current_user.email)
    await service.delete_order(order_id, current_user.id)
    logger.info("Order id=%s deleted", order_id)

Each endpoint passes current_user.id to the service — the controller never fetches the data from database directly. Because user_id always comes from the JWT (never from the request body), a user cannot fetch details of another user's order.

Key Concepts: Async
Async/Await — Handling Requests Efficiently

FastAPI is built on ASGI, supporting asynchronous request handling. Using async/await lets your API serve multiple requests concurrently without blocking — especially for database calls and external HTTP requests.

async def get_orders(self, user_id: int) -> List[Order]:
    result = await self.db.execute(
        select(Order).where(Order.user_id == user_id).order_by(Order.created_at.desc())
    )
    return list(result.scalars().all())

Without async/await, each database call blocks the entire process — no other request can be handled until it finishes. With it, the event loop can serve other requests while waiting for the DB to respond.

Note: Only use async for I/O-bound operations (database, HTTP). CPU-heavy tasks should run in background workers.

Bringing It All Together

#app/main.py
import logging.config
import os
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from starlette.middleware.cors import CORSMiddleware

import app.models  # registers Order + User with Base.metadata
import app.utils.constants as const
from app.api.order_api import router as order_router
from app.api.v1.router import api_router
from app.db.database import create_tables_if_not_exist
from app.middleware import LoggingMiddleware
from app.utils.exceptions import (
    OrderNotFoundException,
    OrderAlreadyExistsException,
    order_not_found_exception_handler,
    order_already_exists_exception_handler,
    validation_exception_handler,
    generic_exception_handler,
)

LOGDIR = const.LOGS_DIR
if not os.path.exists(LOGDIR):
    os.makedirs(LOGDIR, exist_ok=True)

logging.config.fileConfig(const.LOGGING_CONFIG_FILE, disable_existing_loggers=False)
logger = logging.getLogger("fastapi.app")


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Starting up Order Management System")
    await create_tables_if_not_exist()
    logger.info("Database tables verified/created successfully")
    yield
    logger.info("Shutting down Order Management System")


app = FastAPI(
    title="Order Management System",
    description=(
        "## Authentication\n"
        "1. **Register** via `POST /api/v1/auth/register`\n"
        "2. **Login** via `POST /api/v1/auth/login` — use your email as *username*\n"
        "3. Click **Authorize** and paste the token\n"
        "4. All Order endpoints are now unlocked — you will only see **your own** orders"
    ),
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(LoggingMiddleware)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# ── Exception handlers ────────────────────────────────────────────────────────
app.add_exception_handler(OrderNotFoundException,      order_not_found_exception_handler)
app.add_exception_handler(OrderAlreadyExistsException, order_already_exists_exception_handler)
app.add_exception_handler(RequestValidationError,      validation_exception_handler)
app.add_exception_handler(Exception,                   generic_exception_handler)

# ── Routers ───────────────────────────────────────────────────────────────────
app.include_router(api_router)    # /api/v1/auth/... and /api/v1/users/...
app.include_router(order_router)  # /api/orders/...


@app.get("/", tags=["System"])
def welcome():
    return {"message": "Welcome to Order Management System", "docs": "/docs"}


@app.get("/health", tags=["System"])
def health():
    return {"status": "healthy"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

• var/logs/ is created automatically before logging is configured.
• logging.config.fileConfig loads the full setup from config/logging.ini.
• lifespan replaces the deprecated @app.on_event("startup") — startup before yield, shutdown after.
• LoggingMiddleware is registered first so it wraps every request.
• add_exception_handler wires each typed exception to its handler — the service raises, FastAPI catches, the handler responds.

Start it with:

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

Visit http://localhost:8000/docs for interactive Swagger UI — automatically generated from your code.

Swagger UI

Register the user through http://0.0.0.0:8000/api/v1/auth/register endpoint.

Register user

once register , login with the user credentials through http://0.0.0.0:8000/api/v1/auth/login endpoint to get the access token.

Login user

once access token generated , use that to access order api’s through authorisation.

Orders API on successful auth token verification

once authorized , you should be able to perform any crud operations on Order’s API’s.

Order API’s

Database Migrations with Alembic

The problem: create_all() only creates new tables — it never modifies existing ones. Add a column to your model and production breaks.

Alembic fixes this by tracking every schema change as a versioned, reversible script.

pip install alembic
alembic init alembic

#alembic.ini
sqlalchemy.url =

#alembic/env.py - inject URL + handle async
import asyncio
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from app.core.settings import settings
from app.db.database import Base
import app.models  # registers models with Base.metadata
config = context.config
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)
target_metadata = Base.metadata
def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()
async def run_async_migrations():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.", poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()
def run_migrations_online():
    asyncio.run(run_async_migrations())
run_migrations_online()

# Generate migration from model changes
alembic revision --autogenerate -m "add column X"
# Apply
alembic upgrade head
# Rollback
alembic downgrade -1

• Remove create_all() once Alembic is in use — don't mix both
• Run alembic upgrade head in your deploy pipeline before starting the app
• Always import app.models in env.py so Alembic can see your models

Testing

Tests use an in-memory SQLite database so they are fully isolated from the real PostgreSQL DB. An auth_client fixture handles register + login automatically so tests stay clean.

#tests/conftest.py
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.main import app
from app.db.database import Base, get_db

TEST_DB     = "sqlite+aiosqlite:///:memory:"
engine      = create_async_engine(TEST_DB, connect_args={"check_same_thread": False}, poolclass=StaticPool)
TestSession = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def override_get_db():
    async with TestSession() as session:
        yield session

app.dependency_overrides[get_db] = override_get_db


@pytest_asyncio.fixture(autouse=True)
async def setup_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


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


@pytest_asyncio.fixture
async def auth_client(client: AsyncClient):
    """Registers a user, logs in, returns client with Bearer token pre-set."""
    await client.post("/api/v1/auth/register", json={
        "email": "user@example.com", "password": "secret123"
    })
    resp  = await client.post("/api/v1/auth/login", data={
        "username": "user@example.com", "password": "secret123"
    })
    token = resp.json()["access_token"]
    client.headers.update({"Authorization": f"Bearer {token}"})
    return client

#tests/test_orders_api.py
import pytest
from httpx import AsyncClient

ORDER_URL    = "/api/orders/"
SAMPLE_ORDER = {"product_name": "Wireless Mouse", "quantity": 2, "unit_price": 29.99}


@pytest.mark.asyncio
async def test_create_order(auth_client: AsyncClient):
    response = await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)
    assert response.status_code == 201
    assert response.json()["product_name"] == "Wireless Mouse"
    assert response.json()["status"] == "pending"


@pytest.mark.asyncio
async def test_get_my_orders(auth_client: AsyncClient):
    await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)
    response = await auth_client.get(ORDER_URL)
    assert response.status_code == 200
    assert len(response.json()) == 1


@pytest.mark.asyncio
async def test_order_not_found(auth_client: AsyncClient):
    assert (await auth_client.get(f"{ORDER_URL}999")).status_code == 404


@pytest.mark.asyncio
async def test_update_order_status(auth_client: AsyncClient):
    created  = (await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)).json()
    response = await auth_client.put(f"{ORDER_URL}{created['id']}", json={"status": "confirmed"})
    assert response.status_code == 200
    assert response.json()["status"] == "confirmed"


@pytest.mark.asyncio
async def test_delete_order(auth_client: AsyncClient):
    created = (await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)).json()
    assert (await auth_client.delete(f"{ORDER_URL}{created['id']}")).status_code == 204
    assert (await auth_client.get(f"{ORDER_URL}{created['id']}")).status_code == 404


@pytest.mark.asyncio
async def test_duplicate_pending_order_rejected(auth_client: AsyncClient):
    await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)
    response = await auth_client.post(ORDER_URL, json=SAMPLE_ORDER)
    assert response.status_code == 409   # OrderAlreadyExistsException → Conflict


@pytest.mark.asyncio
async def test_unauthenticated_cannot_access_orders(client: AsyncClient):
    assert (await client.get(ORDER_URL)).status_code == 401


@pytest.mark.asyncio
async def test_user_cannot_see_other_users_orders(client: AsyncClient):
    # User A places an order
    await client.post("/api/v1/auth/register", json={"email": "a@test.com", "password": "pass"})
    login_a = await client.post("/api/v1/auth/login", data={"username": "a@test.com", "password": "pass"})
    client.headers.update({"Authorization": f"Bearer {login_a.json()['access_token']}"})
    await client.post(ORDER_URL, json=SAMPLE_ORDER)

    # User B logs in — should see zero orders
    await client.post("/api/v1/auth/register", json={"email": "b@test.com", "password": "pass"})
    login_b = await client.post("/api/v1/auth/login", data={"username": "b@test.com", "password": "pass"})
    client.headers.update({"Authorization": f"Bearer {login_b.json()['access_token']}"})

    assert len((await client.get(ORDER_URL)).json()) == 0

pytest          # run all tests
pytest -v       # verbose output
pytest tests/test_order_api.py -v

Deployment with Docker

The app and PostgreSQL run as separate services. Docker Compose wires them together.

#Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

#docker-compose.yaml
version: '3.9'
services:
  db:
    image: postgres:15
    container_name: orders_db
    environment:
      POSTGRES_DB:       orders
      POSTGRES_USER:     postgres
      POSTGRES_PASSWORD: postgres
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 10s
      timeout: 5s
      retries: 5
  app:
    build: .
    container_name: order_api
    ports:
      - "8000:8000"
    volumes:
      - app_logs:/app/var/logs
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/orders
      - SECRET_KEY=change-me-in-production
    depends_on:
      db:
        condition: service_healthy
    restart: unless-stopped
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 20s
volumes:
  postgres_data:
  app_logs:

docker-compose up --build

The app waits for PostgreSQL to be healthy before starting — the depends_on: condition: service_healthy block handles this automatically.

Production Guidelines

• Don’t mix everything in one file. Routes handle HTTP, services handle business logic, the database module handles sessions. Keep them separate.
• Never trust user input for ownership. The user_id on an order must always come from the JWT token — never from the request body. A user could send any user_id they want in the body.
• Scope all queries by user_id. Every data-access method in the service should filter by user_id. If you forget one endpoint, users can leak each other's data.
• Never store plain passwords. Always hash with bcrypt before persisting. Never log or return password fields.
• Set a strong SECRET_KEY in production. Rotating it invalidates all existing tokens immediately.
• Keep config in one place. Use BaseSettings + .env files per environment. No hardcoded values.
• Never open a database connection without closing it. Use async with AsyncSessionLocal() as session — it closes automatically even on exceptions.
• Raise typed exceptions in the service, not HTTPException. Define custom exception classes once, register handlers in main.py, and keep the service and controller layers free of HTTP concerns. Every layer speaks its own language.
• Never return None or False for errors. Returning None from a service method forces every caller to add a if not result: check. A typed exception propagates cleanly and is handled in one place.
• Write tests early. The data isolation test (user cannot see another user’s orders) is the most important one — it catches the most dangerous class of bug in user-owned data APIs.

You can find the complete working code on GitHub: fastapi

Thank you for reading this article. Please provide your valuable suggestions/ feedback.

• Clap and Share if you liked the content.
• 📰 Read more content on my Medium (on Java Developer interview questions)
• 🔔 Follow me on: LinkedIn
• Reach out to me for resume preparation and interview ready. Contact me here.

Please find my other helpful articles on Java Developer interview questions.

Following are some of the frequently asked Java 8 Interview Questions

Frequently Asked Java Programs

Dear Readers, these are the commonly asked Java programs to check your ability in writing the logic.

SpringBoot Interview Questions | Medium

Rest and Microservices Interview Questions| Medium

Spring Boot tutorial | Medium

Must-know-coding-programs-for-faang-companies| Medium

References:

https://fastapi.tiangolo.com/