Perfect 🎯 — here’s a complete, beginner-friendly, blog-style walkthrough of your FastAPI + Docker Compose project, written like a professional developer article you could publish on Medium, Dev.to, or your own website. It’s structured for education, SEO, and clarity, with code snippets, commands, and clear sections.

🚀 Build Your First FastAPI Project with Docker Compose (Step-by-Step Tutorial)

By Sreeprakash Neelakantan AI, Cloud, and Automation Consultant | NVIDIA-Certified Generative AI Specialist

🌱 Introduction

If you’re a Python developer looking to build modern, high-performance APIs, FastAPI is one of the best frameworks out there. It’s fast, type-safe, easy to learn, and built on top of Starlette and Pydantic, giving you automatic validation, interactive documentation, and blazing speed.

In this tutorial, you’ll build a fully functional FastAPI project that runs with Docker Compose, includes multiple endpoints, file upload/download, background tasks, headers, cookies, authentication, and even WebSockets — all in a single setup.

By the end, you’ll be able to:

✅ Create and run a FastAPI app inside Docker ✅ Explore automatic API docs in your browser ✅ Interact with the app using curl and Swagger UI ✅ Understand middleware, routers, and models ✅ Extend it for your own projects easily

🧩 Prerequisites

Before we start, make sure you have:

🐳 Docker and Docker Compose installed
💻 Python 3.10+ (optional, only if you want to run locally)
🧠 Basic familiarity with Python dictionaries, functions, and JSON

That’s all!

⚙️ Step 1: Project Structure

Here’s how your project will look:

fastapi-docker-compose-tutorial/
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── README.md
└── app/
    ├── main.py
    ├── deps.py
    ├── models.py
    ├── middleware.py
    └── routers/
        ├── health.py
        ├── items.py
        ├── users.py
        ├── misc.py
        └── files.py

Each component plays a specific role:

File	Purpose
`main.py`	Entry point — connects all routers
`routers/`	Contains all endpoint definitions
`models.py`	Defines Pydantic data models
`middleware.py`	Adds timing and logging middleware
`deps.py`	Handles dependencies like API keys and pagination
`Dockerfile`	Builds the app into a Docker image
`docker-compose.yml`	Defines how containers run
`requirements.txt`	Lists dependencies

🧱 Step 2: Writing the Code

Let’s go through the core files that make this work.

🧩 main.py

from fastapi import FastAPI
from .routers import health, items, users, misc, files
from .middleware import TimingMiddleware

app = FastAPI(title="FastAPI Beginner Tutorial", version="1.0.0")

# Add timing middleware
app.add_middleware(TimingMiddleware)

# Register routers
app.include_router(health.router)
app.include_router(misc.router)
app.include_router(items.router)
app.include_router(users.router)
app.include_router(files.router)

This file wires together all your routes and adds a custom middleware that measures response time.

🧠 middleware.py

import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        elapsed = (time.time() - start) * 1000
        response.headers["X-Process-Time-ms"] = f"{elapsed:.2f}"
        return response

Adds an X-Process-Time-ms header showing how long each request took.

📦 Dockerfile

FROM python:3.11-slim
WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

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

COPY app ./app
EXPOSE 8000

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

🐳 docker-compose.yml

services:
  api:
    build: .
    container_name: fastapi_tutorial_api
    ports:
      - "8000:8000"
    volumes:
      - ./app:/app/app:rw
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

This allows live code reload when editing files locally.

📜 requirements.txt

fastapi==0.115.2
uvicorn[standard]==0.31.1
pydantic==2.9.2
python-multipart==0.0.12
email-validator==2.2.0

🚀 Step 3: Run the App

Open your terminal and run:

docker compose up --build

Then visit these URLs:

You should see FastAPI’s auto-generated documentation.

🧪 Step 4: Play With Endpoints

Here are a few ready-to-try commands:

Health

curl -s http://localhost:8000/health

Items CRUD

# Create
curl -s -X POST http://localhost:8000/items -H "Content-Type: application/json" \
  -d '{"name":"USB Hub","price":1299,"tags":["usb","hub"]}'
# List
curl -s http://localhost:8000/items
# Get
curl -s http://localhost:8000/items/1
# Update
curl -s -X PUT http://localhost:8000/items/1 -H "Content-Type: application/json" \
  -d '{"name":"USB Hub Pro","price":1499,"tags":["usb","hub","pro"]}'
# Delete
curl -s -X DELETE http://localhost:8000/items/1

Users

curl -s -X POST http://localhost:8000/users/register \
  -H "Content-Type: application/json" \
  -d '{"email":"test@example.com","password":"secret"}'

File Upload

curl -s -F "file=@README.md" http://localhost:8000/files/upload

Background Task

curl -s "http://localhost:8000/background/send-email?to=demo@example.com"

WebSocket Echo

wscat -c ws://localhost:8000/ws/echo
# Type: hello
# Response: echo: hello

🔍 Step 5: Explore in Swagger UI

Visit http://localhost:8000/docs. You can “Try it out” right from your browser — no external tools required.

Swagger automatically documents:

Query parameters
Path variables
Request bodies
Response schemas
Status codes

🧠 Step 6: Understand the Core Concepts

Concept	Explanation
Routers	Organize endpoints into logical modules (e.g., `/items`, `/users`)
Models	Define input/output schemas using Pydantic
Middleware	Modify requests/responses globally (e.g., add headers, timing)
Dependencies	Reusable logic like pagination or API key verification
Background Tasks	Run async jobs without blocking requests
WebSockets	Enable real-time communication
Swagger UI	Auto-generated docs and testing interface

💡 Step 7: Extend Your Project

Once you understand this template, you can easily add:

Database integration (SQLite or PostgreSQL)
JWT-based authentication
File storage with AWS S3 or local volume
Custom error handlers
Unit tests using pytest and httpx

🧾 Quick Recap

Step	Action
1	Create project structure
2	Write FastAPI app and routers
3	Add middleware and models
4	Set up Docker and Compose
5	Run with `docker compose up`
6	Test endpoints via Swagger and curl
7	Extend and scale

🎯 Why This Matters

By containerizing your FastAPI app early, you:

Ensure consistent environments across machines
Avoid dependency issues (“it works on my machine!”)
Make deployment to cloud (AWS, Azure, GCP) effortless
Learn the DevOps mindset from day one

This small setup mirrors how real-world production APIs are structured — clean, modular, and easily testable.

✨ Final Thoughts

FastAPI + Docker Compose is a perfect combo for modern backend development. It gives you:

Type safety
Speed
Simplicity
Instant documentation

Whether you’re building a startup MVP, an internal tool, or a full SaaS backend, this foundation will scale with you.

💬 Your Turn

Clone it, run it, and explore:

Download the full project ZIP

Then tweak it — add authentication, connect a database, or build your own mini API service.

Author: Sreeprakash Neelakantan Founder & Managing Director, Schogini Systems Private Limited NVIDIA-Certified Associate in Generative AI LLMs

Helping small businesses and professionals adopt AI, automation, and modern development stacks for real-world growth.

Part 2: Adding a Database and JWT Authentication to This FastAPI Docker Project

Perfect — let’s take this project to the next level 🚀

Here’s Part 2 of the blog series, expanding your FastAPI + Docker Compose foundation to include a real database (SQLite) and JWT-based authentication — exactly how production-grade APIs handle users securely.

🧱 Part 2: Adding a Database and JWT Authentication to Your FastAPI + Docker Project

By Sreeprakash Neelakantan NVIDIA-Certified AI Professional | Founder of Schogini Systems Private Limited

🎯 What You’ll Build

We’ll extend the previous project with:

✅ A SQLite database (lightweight but production-ready structure)
✅ SQLAlchemy ORM for easy table mapping
✅ User registration and login stored in the database
✅ JWT (JSON Web Tokens) for secure authentication
✅ Protected endpoints using JWT Bearer tokens

This makes your FastAPI app a true backend skeleton — ready for real users.

🧩 Step 1 – Add New Dependencies

Update your requirements.txt to include:

fastapi==0.115.2
uvicorn[standard]==0.31.1
pydantic==2.9.2
python-multipart==0.0.12
email-validator==2.2.0
SQLAlchemy==2.0.36
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0

Rebuild your container:

docker compose build
docker compose up

🗄️ Step 2 – Add Database Setup (`app/db.py`)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

👥 Step 3 – Define User Model (`app/models_db.py`)

from sqlalchemy import Column, Integer, String
from .db import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

Initialize the database on startup (edit main.py):

from .db import Base, engine
from .models_db import User

@app.on_event("startup")
def on_startup():
    Base.metadata.create_all(bind=engine)

🔐 Step 4 – JWT Utility (`app/auth.py`)

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from .db import get_db
from .models_db import User

SECRET_KEY = "supersecretkey-change-this"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

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

def verify_password(plain, hashed): return pwd_context.verify(plain, hashed)
def get_password_hash(pw): return pwd_context.hash(pw)

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

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    cred_exc = 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])
        email: str = payload.get("sub")
        if email is None:
            raise cred_exc
    except JWTError:
        raise cred_exc
    user = db.query(User).filter(User.email == email).first()
    if user is None:
        raise cred_exc
    return user

🧑‍💻 Step 5 – Auth Router (`app/routers/auth.py`)

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from ..db import get_db
from ..models_db import User
from ..auth import create_access_token, get_password_hash, verify_password

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register")
def register(email: str, password: str, db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(status_code=400, detail="Email already registered")
    user = User(email=email, hashed_password=get_password_hash(password))
    db.add(user)
    db.commit()
    db.refresh(user)
    return {"message": "User registered", "email": user.email}

@router.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token(data={"sub": user.email})
    return {"access_token": token, "token_type": "bearer"}

🧰 Step 6 – Protect Routes (`app/routers/secure.py`)

from fastapi import APIRouter, Depends
from ..auth import get_current_user
from ..models_db import User

router = APIRouter(prefix="/secure", tags=["secure"])

@router.get("/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return {"email": current_user.email}

Include it in main.py:

from .routers import secure, auth
app.include_router(auth.router)
app.include_router(secure.router)

🧪 Step 7 – Try It Out

Register a User

curl -X POST "http://localhost:8000/auth/register?email=demo@example.com&password=secret"

Login to Get Token

curl -X POST -d "username=demo@example.com&password=secret" \
     -H "Content-Type: application/x-www-form-urlencoded" \
     http://localhost:8000/auth/token

Response:

{"access_token": "eyJhbGciOi...", "token_type": "bearer"}

Access a Protected Route

curl -H "Authorization: Bearer eyJhbGciOi..." \
     http://localhost:8000/secure/me

You’ll get your email back only if the token is valid.

🔍 How It Works

Register: password → bcrypt hash → stored in SQLite
Login: credentials → token with JWT payload (sub=email)
Protected Routes: FastAPI depends on get_current_user to validate token

🧠 Concepts You Learned

Concept	Explanation
SQLAlchemy	ORM for managing database tables via Python classes
Passlib	Safely hash passwords with bcrypt
JWT	Compact, signed token that authenticates API calls
OAuth2PasswordBearer	FastAPI helper to extract `Authorization` header tokens
Dependency Injection	Makes auth logic reusable via `Depends()`

🧾 Quick Recap

Step	Action
1	Added SQLAlchemy, Passlib, and JWT dependencies
2	Configured SQLite database
3	Created User model
4	Built JWT auth helpers
5	Added register and login routes
6	Protected API routes with Bearer token
7	Tested everything with curl and Swagger UI

💡 Next Steps (Part 3 Preview)

Add PostgreSQL as a service in Docker Compose
Implement JWT refresh tokens
Use SQLAlchemy Async ORM
Write Pytest tests for auth and CRUD flows
Integrate Redis for token blacklisting or rate limiting

✨ Final Thoughts

You’ve now transformed a simple FastAPI demo into a complete backend foundation with:

Secure user auth
Persistent data storage
Scalable architecture with Docker

From here, you can plug in a frontend (React, Next.js, or Svelte), add more models, and deploy on AWS or DigitalOcean.

Author: Sreeprakash Neelakantan NVIDIA-Certified Associate in Generative AI LLMs | Founder & MD, Schogini Systems Private Limited

Helping small businesses and professionals leverage AI and automation for 24×7 growth.

Perfect—here’s Part 3 of your series. We’ll Dockerize PostgreSQL, switch your project to Async SQLAlchemy, and add a pytest test-suite (with async HTTP testing). It’s practical, end-to-end, and slots neatly onto your Part 1/2 codebase.

🐘 Part 3: Dockerizing PostgreSQL + Async SQLAlchemy ORM + Pytest Setup

By Sreeprakash Neelakantan NVIDIA-Certified AI | Founder & MD, Schogini Systems Private Limited

🎯 Goals

Add PostgreSQL as a Docker service
Migrate your app to Async SQLAlchemy (with asyncpg)
Keep your JWT auth from Part 2
Ship a pytest suite using httpx.AsyncClient + pytest-asyncio

You’ll get a production-flavored stack while staying beginner-friendly.

✅ What We’ll Add/Change

New/updated dependencies (requirements.txt):

fastapi==0.115.2
uvicorn[standard]==0.31.1
pydantic==2.9.2
python-multipart==0.0.12
email-validator==2.2.0
SQLAlchemy==2.0.36
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
asyncpg==0.29.0
pytest==8.3.3
pytest-asyncio==0.24.0
httpx==0.27.2
asgi-lifespan==2.1.0

Rebuild after editing:

docker compose build

🐳 1) Add PostgreSQL to docker-compose

Update your docker-compose.yml:

services:
  api:
    build: .
    container_name: fastapi_tutorial_api
    ports:
      - "8000:8000"
    volumes:
      - ./app:/app/app:rw
    environment:
      - DATABASE_URL=postgresql+asyncpg://app:app@db:5432/app
      - LOG_LEVEL=info
      - JWT_SECRET=change-me
    depends_on:
      db:
        condition: service_healthy
    command: >
      sh -c "uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload"

  db:
    image: postgres:16-alpine
    container_name: fastapi_tutorial_db
    environment:
      - POSTGRES_DB=app
      - POSTGRES_USER=app
      - POSTGRES_PASSWORD=app
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U app -d app']
      interval: 3s
      timeout: 3s
      retries: 10

volumes:
  pgdata:

Why depends_on.healthcheck? It ensures the API waits until Postgres is ready.

Bring both up:

docker compose up --build

🧠 2) Async SQLAlchemy Setup

Create app/db_async.py:

# app/db_async.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
import os

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://app:app@db:5432/app")

engine = create_async_engine(DATABASE_URL, echo=False, future=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with SessionLocal() as session:
        yield session

🧱 3) Define Async Models

Replace your Part-2 sync models with async-friendly SQLAlchemy models.

app/models_db.py

# app/models_db.py
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer, Float
from .db_async import Base

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String)

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String, index=True)
    price: Mapped[float] = mapped_column(Float)
    tags: Mapped[str] = mapped_column(String, default="")  # comma-separated for demo

We keep items simple: tags as a comma-separated string (beginner-friendly). You can normalize later.

🚀 4) Create Tables on Startup

Edit app/main.py to create tables with async engine:

# app/main.py
from fastapi import FastAPI
from .routers import health, items, users, misc, files
from .middleware import TimingMiddleware
from .db_async import Base, engine
from .models_db import User, Item

app = FastAPI(title="FastAPI Beginner Tutorial", version="3.0.0")
app.add_middleware(TimingMiddleware)

@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# routers
app.include_router(health.router)
app.include_router(misc.router)
app.include_router(items.router)   # we'll update to async DB
app.include_router(users.router)   # we'll update to async DB
app.include_router(files.router)

🔐 5) Update Auth Helpers (Async DB)

app/auth.py (only changes: import async Session and query with await):

# app/auth.py
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from .db_async import get_db
from .models_db import User
import os

SECRET_KEY = os.getenv("JWT_SECRET", "change-me")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

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

def verify_password(plain, hashed): return pwd_context.verify(plain, hashed)
def get_password_hash(pw): return pwd_context.hash(pw)

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

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
    cred_exc = 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])
        email: str = payload.get("sub")
        if email is None:
            raise cred_exc
    except JWTError:
        raise cred_exc

    result = await db.execute(select(User).where(User.email == email))
    user = result.scalar_one_or_none()
    if user is None:
        raise cred_exc
    return user

👤 6) Auth Router (Async)

app/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from ..db_async import get_db
from ..models_db import User
from ..auth import create_access_token, get_password_hash, verify_password

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register")
async def register(email: str, password: str, db: AsyncSession = Depends(get_db)):
    exists = await db.execute(select(User).where(User.email == email))
    if exists.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")
    user = User(email=email, hashed_password=get_password_hash(password))
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return {"message": "User registered", "email": user.email}

@router.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(User).where(User.email == form.username))
    user = res.scalar_one_or_none()
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token({"sub": user.email})
    return {"access_token": token, "token_type": "bearer"}

from .routers import auth, secure
app.include_router(auth.router)
app.include_router(secure.router)

app/routers/secure.py

from fastapi import APIRouter, Depends
from ..auth import get_current_user
from ..models_db import User

router = APIRouter(prefix="/secure", tags=["secure"])

@router.get("/me")
async def read_me(current_user: User = Depends(get_current_user)):
    return {"email": current_user.email}

📦 7) Items Router using Async DB

app/routers/items.py (now DB-backed; replaces in-memory):

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete
from typing import Optional, List
from ..db_async import get_db
from ..models_db import Item
from ..models import ItemIn, Item as ItemOut

router = APIRouter(prefix="/items", tags=["items"])

def _tags_to_str(tags: list[str]) -> str:
    return ",".join(tags)

def _tags_to_list(s: str) -> list[str]:
    return [t for t in (s or "").split(",") if t]

@router.post("", response_model=ItemOut, status_code=201)
async def create_item(payload: ItemIn, db: AsyncSession = Depends(get_db)):
    obj = Item(name=payload.name, price=payload.price, tags=_tags_to_str(payload.tags))
    db.add(obj)
    await db.commit()
    await db.refresh(obj)
    return ItemOut(id=obj.id, name=obj.name, price=obj.price, tags=_tags_to_list(obj.tags))

@router.get("", response_model=List[ItemOut])
async def list_items(skip: int = 0, limit: int = 10, tag: Optional[str] = None, db: AsyncSession = Depends(get_db)):
    stmt = select(Item).offset(skip).limit(limit)
    if tag:
        # naive filter: tag substring match
        stmt = select(Item).where(Item.tags.like(f"%{tag}%")).offset(skip).limit(limit)
    res = await db.execute(stmt)
    rows = res.scalars().all()
    return [ItemOut(id=i.id, name=i.name, price=i.price, tags=_tags_to_list(i.tags)) for i in rows]

@router.get("/{item_id}", response_model=ItemOut)
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(Item).where(Item.id == item_id))
    obj = res.scalar_one_or_none()
    if not obj:
        raise HTTPException(status_code=404, detail="Item not found")
    return ItemOut(id=obj.id, name=obj.name, price=obj.price, tags=_tags_to_list(obj.tags))

@router.put("/{item_id}", response_model=ItemOut)
async def update_item(item_id: int, payload: ItemIn, db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(Item).where(Item.id == item_id))
    obj = res.scalar_one_or_none()
    if not obj:
        raise HTTPException(status_code=404, detail="Item not found")
    obj.name = payload.name
    obj.price = payload.price
    obj.tags = _tags_to_str(payload.tags)
    await db.commit()
    await db.refresh(obj)
    return ItemOut(id=obj.id, name=obj.name, price=obj.price, tags=_tags_to_list(obj.tags))

@router.delete("/{item_id}", status_code=204)
async def delete_item(item_id: int, db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(Item).where(Item.id == item_id))
    obj = res.scalar_one_or_none()
    if not obj:
        raise HTTPException(status_code=404, detail="Item not found")
    await db.execute(delete(Item).where(Item.id == item_id))
    await db.commit()
    return None

🧪 8) Pytest: Async API Tests

Create a tests folder:

app/
  ...
tests/
  conftest.py
  test_auth.py
  test_items.py

tests/conftest.py – start the FastAPI app in-process and use an async client.

import os
import pytest
import asyncio
from httpx import AsyncClient
from asgi_lifespan import LifespanManager
from app.main import app

# Use a separate DB for tests if you like, e.g. "app_test"
# os.environ["DATABASE_URL"] = "postgresql+asyncpg://app:app@db:5432/app_test"

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop

@pytest.fixture(scope="session", async_fixture=True)
async def test_client():
    async with LifespanManager(app):
        async with AsyncClient(app=app, base_url="http://testserver") as ac:
            yield ac

tests/test_auth.py

import pytest

@pytest.mark.asyncio
async def test_register_and_login(test_client):
    # register
    r = await test_client.post("/auth/register", params={"email": "t@e.com", "password": "secret"})
    assert r.status_code in (200, 400)  # allow "already registered" on re-run

    # login
    r = await test_client.post("/auth/token", data={"username": "t@e.com", "password": "secret"})
    assert r.status_code == 200
    token = r.json()["access_token"]
    assert token

    # /secure/me
    r = await test_client.get("/secure/me", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert r.json()["email"] == "t@e.com"

tests/test_items.py

import pytest

@pytest.mark.asyncio
async def test_items_crud(test_client):
    payload = {"name": "Pen", "price": 9.99, "tags": ["stationery"]}
    r = await test_client.post("/items", json=payload)
    assert r.status_code == 201
    item = r.json()
    item_id = item["id"]

    r = await test_client.get(f"/items/{item_id}")
    assert r.status_code == 200
    assert r.json()["name"] == "Pen"

    r = await test_client.put(f"/items/{item_id}", json={"name":"Pen Pro","price":12.5,"tags":["stationery","pro"]})
    assert r.status_code == 200
    assert r.json()["price"] == 12.5

    r = await test_client.get("/items?tag=pro")
    assert r.status_code == 200
    assert any(i["id"] == item_id for i in r.json())

    r = await test_client.delete(f"/items/{item_id}")
    assert r.status_code == 204

    r = await test_client.get(f"/items/{item_id}")
    assert r.status_code == 404

Run tests (inside the api container or on host with env pointing to db):

# If running inside the API container shell:
docker compose exec api bash -lc "pytest -q"

# Or from host (tests use in-process app; DB must be running):
pytest -q

Tip: For completely isolated test DBs, create a app_test database and set DATABASE_URL in conftest.py before importing the app (shown as a comment).

🧪 Quick Sanity Runs (cURL)

# Health
curl -s http://localhost:8000/health

# Register + Login
curl -s -X POST "http://localhost:8000/auth/register?email=demo@example.com&password=secret"
curl -s -X POST -H "Content-Type: application/x-www-form-urlencoded" \
     -d "username=demo@example.com&password=secret" \
     http://localhost:8000/auth/token

# Use token
TOKEN="paste_token_here"
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/secure/me

# Items CRUD
curl -s -X POST http://localhost:8000/items -H "Content-Type: application/json" \
  -d '{"name":"Book","price":199.0,"tags":["reading"]}'
curl -s http://localhost:8000/items

🧠 What You Learned

Postgres in Docker with health checks & volumes
Async SQLAlchemy with create_async_engine, async_sessionmaker, select()
JWT auth across async endpoints
Async testing with httpx.AsyncClient + pytest-asyncio + asgi-lifespan

🪜 Next (Part 4) Ideas

Add Alembic migrations (versioned schema changes)
Switch to Async relationships and normalized tags
Add role-based authorization (admin/user)
Introduce Redis for rate limiting or token blacklist
Wire GitHub Actions CI to run tests on every push

🚀 Build Your First FastAPI Project with Docker Compose (Step-by-Step Tutorial)

🌱 Introduction

🧩 Prerequisites

⚙️ Step 1: Project Structure

🧱 Step 2: Writing the Code

🧩 main.py

🧠 middleware.py

📦 Dockerfile

🐳 docker-compose.yml

📜 requirements.txt

🚀 Step 3: Run the App

🧪 Step 4: Play With Endpoints

Health

Items CRUD

Users

File Upload

Background Task

WebSocket Echo

🔍 Step 5: Explore in Swagger UI

🧠 Step 6: Understand the Core Concepts

💡 Step 7: Extend Your Project

🧾 Quick Recap

🎯 Why This Matters

✨ Final Thoughts

💬 Your Turn

Part 2: Adding a Database and JWT Authentication to This FastAPI Docker Project

🧱 Part 2: Adding a Database and JWT Authentication to Your FastAPI + Docker Project

🎯 What You’ll Build

🧩 Step 1 – Add New Dependencies

🗄️ Step 2 – Add Database Setup (app/db.py)

👥 Step 3 – Define User Model (app/models_db.py)

🔐 Step 4 – JWT Utility (app/auth.py)

🧑‍💻 Step 5 – Auth Router (app/routers/auth.py)

🧰 Step 6 – Protect Routes (app/routers/secure.py)

🧪 Step 7 – Try It Out

Register a User

Login to Get Token

Access a Protected Route

🔍 How It Works

🧠 Concepts You Learned

🧾 Quick Recap

💡 Next Steps (Part 3 Preview)

✨ Final Thoughts

🐘 Part 3: Dockerizing PostgreSQL + Async SQLAlchemy ORM + Pytest Setup

🎯 Goals

✅ What We’ll Add/Change

🐳 1) Add PostgreSQL to docker-compose

🧠 2) Async SQLAlchemy Setup

🧱 3) Define Async Models

🚀 4) Create Tables on Startup

🔐 5) Update Auth Helpers (Async DB)

👤 6) Auth Router (Async)

📦 7) Items Router using Async DB

🧪 8) Pytest: Async API Tests

🧪 Quick Sanity Runs (cURL)

🧠 What You Learned

🪜 Next (Part 4) Ideas

🗄️ Step 2 – Add Database Setup (`app/db.py`)

👥 Step 3 – Define User Model (`app/models_db.py`)

🔐 Step 4 – JWT Utility (`app/auth.py`)

🧑‍💻 Step 5 – Auth Router (`app/routers/auth.py`)

🧰 Step 6 – Protect Routes (`app/routers/secure.py`)