Python Microservices Tutorial — FastAPI + RabbitMQ + Docker Project

The Story That Explains Microservices

A microservice is a small, independently deployable service that owns one business capability — users, products, payments — and talks to other services through well-defined APIs or messages. Each service has its own database, its own codebase, its own deploy cycle. Together they form a system that scales, fails, and evolves in pieces.

Monolith vs Microservices — Side by Side

The Project — E-commerce Backend

We'll build a small but realistic e-commerce backend called ShopKart. It has four independent Python services that together let a customer register, browse products, and place an order. Each service is a real, runnable FastAPI app with its own SQLite database and its own Dockerfile.

The Architecture Diagram

Before writing a single line of code we draw the picture. This diagram is the contract — every arrow is a network call, every box is a deployable unit.

                         📱  Client (Web / Mobile)
                                  ↓   HTTPS
                    ┌───────────────────────────────────────┐
                    │         🔑  API GATEWAY  (:8000)              │
                    │   JWT auth • routing • rate-limit • CORS      │
                    └─────┌─────────────┌─────────────┌─────────┘
                           │             │             │
                     HTTP  │       HTTP   │      HTTP   │
                           ↓             ↓             ↓
                ┌───────────┐   ┌───────────┐   ┌───────────┐
                │   USER   │   │ PRODUCT  │   │  ORDER   │
                │  :8001   │   │  :8002   │   │  :8003   │
                │ users.db │   │ prods.db │   │ ords.db  │
                └───────────┘   └───────────┘   └─────┌─────┘
                                       ↑              │
                                       └── check stock ─┘
                                                      │
                                          publish     ↓
                                    ┌───────────────────────────┐
                                    │   📧 RabbitMQ Broker   │
                                    │ queue: order.created  │
                                    └───────────┌───────────────┘
                                                │  consume
                                                ↓
                                      ┌────────────────────┐
                                      │  📡 Notification    │
                                      │     Worker             │
                                      └────────────────────┘

Project Structure

# shopkart/
shopkart/
├── docker-compose.yml
├── gateway/
│   ├── main.py
│   ├── requirements.txt
│   └── Dockerfile
├── user_service/
│   ├── main.py
│   ├── models.py
│   ├── requirements.txt
│   └── Dockerfile
├── product_service/
│   ├── main.py
│   ├── models.py
│   ├── requirements.txt
│   └── Dockerfile
├── order_service/
│   ├── main.py
│   ├── models.py
│   ├── broker.py
│   ├── requirements.txt
│   └── Dockerfile
└── notification_worker/
    ├── worker.py
    ├── requirements.txt
    └── Dockerfile

Service 1 — User Service (FastAPI + JWT)

The User Service is our simplest — it registers users, verifies passwords and issues JWTs. Any other service that wants to know "is this token valid?" calls us.

user_service/models.py

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
 
engine = create_engine("sqlite:///users.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False)
Base = declarative_base()
 
class User(Base):
    __tablename__ = "users"
    id            = Column(Integer, primary_key=True)
    email         = Column(String, unique=True, index=True, nullable=False)
    hashed_pw     = Column(String, nullable=False)
    full_name     = Column(String)
 
Base.metadata.create_all(engine)

user_service/main.py

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr
from passlib.hash import bcrypt
from jose import jwt
from datetime import datetime, timedelta
from models import User, SessionLocal
 
SECRET  = "change-me-in-prod"
ALGO    = "HS256"
app     = FastAPI(title="User Service")
 
class RegisterIn(BaseModel):
    email: EmailStr
    password: str
    full_name: str
 
class LoginIn(BaseModel):
    email: EmailStr
    password: str
 
def get_db():
    db = SessionLocal()
    try: yield db
    finally: db.close()
 
# ------------------------------------------------------- register
@app.post("/register")
def register(payload: RegisterIn, db=Depends(get_db)):
    if db.query(User).filter_by(email=payload.email).first():
        raise HTTPException(400, "Email already registered")
    u = User(
        email=payload.email,
        hashed_pw=bcrypt.hash(payload.password),
        full_name=payload.full_name,
    )
    db.add(u); db.commit(); db.refresh(u)
    return {"id": u.id, "email": u.email}
 
# ------------------------------------------------------- login
@app.post("/login")
def login(payload: LoginIn, db=Depends(get_db)):
    u = db.query(User).filter_by(email=payload.email).first()
    if not u or not bcrypt.verify(payload.password, u.hashed_pw):
        raise HTTPException(401, "Invalid credentials")
    token = jwt.encode(
        {"sub": u.id, "email": u.email,
         "exp": datetime.utcnow() + timedelta(hours=2)},
        SECRET, algorithm=ALGO)
    return {"access_token": token, "token_type": "bearer"}
 
# ------------------------------------------------------- verify (for gateway)
@app.get("/verify/{token}")
def verify(token: str):
    try:
        data = jwt.decode(token, SECRET, algorithms=[ALGO])
        return {"valid": True, "user_id": data["sub"]}
    except Exception:
        raise HTTPException(401, "Invalid or expired token")
 
@app.get("/health")
def health(): return {"status": "ok", "service": "user"}

Service 2 — Product Service

The Product Service is stateless-ish: it holds the catalog, exposes read endpoints, and has one authoritative reserve endpoint that the Order Service calls before confirming a sale. Notice how it never talks to the users database directly.

# product_service/main.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import Column, Integer, String, Float, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
 
engine = create_engine("sqlite:///products.db", connect_args={"check_same_thread": False})
Session = sessionmaker(bind=engine)
Base = declarative_base()
 
class Product(Base):
    __tablename__ = "products"
    id    = Column(Integer, primary_key=True)
    name  = Column(String, nullable=False)
    price = Column(Float,  nullable=False)
    stock = Column(Integer, default=0)
 
Base.metadata.create_all(engine)
app = FastAPI(title="Product Service")
 
class ReserveIn(BaseModel):
    product_id: int
    quantity:   int
 
def db():
    s = Session()
    try: yield s
    finally: s.close()
 
@app.get("/products")
def list_products(s=Depends(db)):
    return [{"id": p.id, "name": p.name,
             "price": p.price, "stock": p.stock}
            for p in s.query(Product).all()]
 
@app.get("/products/{pid}")
def detail(pid: int, s=Depends(db)):
    p = s.get(Product, pid)
    if not p: raise HTTPException(404, "Product not found")
    return {"id": p.id, "name": p.name,
            "price": p.price, "stock": p.stock}
 
# ------------------------- called BY the Order Service -----------
@app.post("/reserve")
def reserve(payload: ReserveIn, s=Depends(db)):
    p = s.get(Product, payload.product_id)
    if not p:
        raise HTTPException(404, "Product not found")
    if p.stock < payload.quantity:
        raise HTTPException(409, "Insufficient stock")
    p.stock -= payload.quantity
    s.commit()
    return {"reserved": payload.quantity,
            "remaining_stock": p.stock,
            "unit_price": p.price}
 
@app.get("/health")
def health(): return {"status": "ok", "service": "product"}

Service 3 — Order Service (HTTP + RabbitMQ)

Order is where the two communication styles meet. It synchronously calls Product to reserve stock, then asynchronously publishes an event so downstream workers can react without slowing down the customer.

order_service/broker.py — reusable RabbitMQ publisher

import json, os, pika
 
RABBIT_URL = os.getenv("RABBIT_URL", "amqp://guest:guest@rabbitmq:5672/")
 
def publish(queue: str, message: dict):
    """Fire-and-forget publish. Creates queue if missing."""
    conn = pika.BlockingConnection(pika.URLParameters(RABBIT_URL))
    ch   = conn.channel()
    ch.queue_declare(queue=queue, durable=True)
    ch.basic_publish(
        exchange="",
        routing_key=queue,
        body=json.dumps(message).encode(),
        properties=pika.BasicProperties(delivery_mode=2)  # persistent
    )
    conn.close()

order_service/main.py

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import httpx, os
from sqlalchemy import Column, Integer, Float, String, DateTime, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime
from broker import publish
 
PRODUCT_URL = os.getenv("PRODUCT_URL", "http://product:8002")
 
engine  = create_engine("sqlite:///orders.db", connect_args={"check_same_thread": False})
Session = sessionmaker(bind=engine)
Base    = declarative_base()
 
class Order(Base):
    __tablename__ = "orders"
    id         = Column(Integer, primary_key=True)
    user_id    = Column(Integer, nullable=False)
    product_id = Column(Integer, nullable=False)
    quantity   = Column(Integer, nullable=False)
    total      = Column(Float,   nullable=False)
    status     = Column(String,  default="CREATED")
    created_at = Column(DateTime, default=datetime.utcnow)
 
Base.metadata.create_all(engine)
app = FastAPI(title="Order Service")
 
class OrderIn(BaseModel):
    user_id:    int
    product_id: int
    quantity:   int
 
def db():
    s = Session()
    try: yield s
    finally: s.close()
 
@app.post("/orders")
def create_order(payload: OrderIn, s=Depends(db)):
    # 1) SYNCHRONOUS call to Product Service
    try:
        r = httpx.post(f"{PRODUCT_URL}/reserve",
                      json={"product_id": payload.product_id,
                            "quantity":   payload.quantity},
                      timeout=3.0)
        r.raise_for_status()
    except httpx.HTTPStatusError as e:
        raise HTTPException(e.response.status_code, e.response.json()["detail"])
    except httpx.RequestError:
        raise HTTPException(503, "Product service unreachable")
 
    unit_price = r.json()["unit_price"]
 
    # 2) Persist the order in OUR OWN database
    order = Order(user_id=payload.user_id,
                  product_id=payload.product_id,
                  quantity=payload.quantity,
                  total=unit_price * payload.quantity)
    s.add(order); s.commit(); s.refresh(order)
 
    # 3) ASYNC event -> whoever cares (email, invoicing, analytics)
    publish("order.created", {
        "order_id":   order.id,
        "user_id":    order.user_id,
        "product_id": order.product_id,
        "quantity":   order.quantity,
        "total":      order.total,
    })
 
    return {"order_id": order.id, "total": order.total,
            "status": order.status}
 
@app.get("/orders/{oid}")
def get_order(oid: int, s=Depends(db)):
    o = s.get(Order, oid)
    if not o: raise HTTPException(404, "Order not found")
    return {"id": o.id, "user_id": o.user_id,
            "total": o.total, "status": o.status}
 
@app.get("/health")
def health(): return {"status": "ok", "service": "order"}

Service 4 — API Gateway

Clients should never call our internal services directly. The gateway is a thin reverse proxy that validates the JWT once, then forwards. Everything sensitive (which port Product runs on, whether we use Redis) stays hidden behind it.

# gateway/main.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import httpx, os
 
app = FastAPI(title="API Gateway")
 
ROUTES = {
    "user":    os.getenv("USER_URL",    "http://user:8001"),
    "product": os.getenv("PRODUCT_URL", "http://product:8002"),
    "order":   os.getenv("ORDER_URL",   "http://order:8003"),
}
 
PUBLIC = {"/user/register", "/user/login", "/health"}
 
async def verify_token(token: str) -> int:
    async with httpx.AsyncClient(timeout=2.0) as c:
        r = await c.get(f"{ROUTES['user']}/verify/{token}")
    if r.status_code != 200:
        raise HTTPException(401, "Invalid token")
    return r.json()["user_id"]
 
@app.api_route("/{service}/{path:path}",
                methods=["GET","POST","PUT","DELETE"])
async def proxy(service: str, path: str, request: Request):
    if service not in ROUTES:
        raise HTTPException(404, "Unknown service")
 
    full_path = f"/{service}/{path}"
    if full_path not in PUBLIC:
        auth = request.headers.get("authorization", "")
        if not auth.startswith("Bearer "):
            raise HTTPException(401, "Missing bearer token")
        await verify_token(auth.split()[1])
 
    target = f"{ROUTES[service]}/{path}"
    body   = await request.body()
 
    async with httpx.AsyncClient(timeout=5.0) as c:
        r = await c.request(request.method, target,
                            content=body, params=request.query_params)
    return JSONResponse(status_code=r.status_code, content=r.json())
 
@app.get("/health")
def health(): return {"status": "ok", "service": "gateway"}

Notification Worker — the Async Consumer

Not every service is a web server. The notification worker has no HTTP port — it just listens to RabbitMQ and reacts. This decouples "an order happened" from "who cares that an order happened".

# notification_worker/worker.py
import json, os, time, pika
 
RABBIT_URL = os.getenv("RABBIT_URL", "amqp://guest:guest@rabbitmq:5672/")
 
def handle_order(ch, method, _props, body):
    event = json.loads(body)
    print(f"[EMAIL] Order #{event['order_id']} confirmed "
          f"for user {event['user_id']} - total ${event['total']}")
    # TODO: send real email here
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
def main():
    while True:
        try:
            conn = pika.BlockingConnection(pika.URLParameters(RABBIT_URL))
            ch   = conn.channel()
            ch.queue_declare(queue="order.created", durable=True)
            ch.basic_qos(prefetch_count=1)
            ch.basic_consume(queue="order.created",
                              on_message_callback=handle_order)
            print("[worker] waiting for messages...")
            ch.start_consuming()
        except pika.exceptions.AMQPConnectionError:
            print("[worker] rabbit down, retry in 3s")
            time.sleep(3)
 
if __name__ == "__main__":
    main()

Dockerfile Template & docker-compose.yml

Every service uses the exact same slim Dockerfile — only the port and the entry file differ. This uniformity is the point.

Dockerfile — reused across services

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

docker-compose.yml — one command to run everything

version: "3.9"
 
services:
 
  rabbitmq:
    image: rabbitmq:3-management
    ports: ["5672:5672", "15672:15672"]
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 10s
 
  user:
    build: ./user_service
    command: uvicorn main:app --host 0.0.0.0 --port 8001
    ports: ["8001:8001"]
 
  product:
    build: ./product_service
    command: uvicorn main:app --host 0.0.0.0 --port 8002
    ports: ["8002:8002"]
 
  order:
    build: ./order_service
    command: uvicorn main:app --host 0.0.0.0 --port 8003
    environment:
      PRODUCT_URL: http://product:8002
      RABBIT_URL:  amqp://guest:guest@rabbitmq:5672/
    ports: ["8003:8003"]
    depends_on: [product, rabbitmq]
 
  gateway:
    build: ./gateway
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    environment:
      USER_URL:    http://user:8001
      PRODUCT_URL: http://product:8002
      ORDER_URL:   http://order:8003
    ports: ["8000:8000"]
    depends_on: [user, product, order]
 
  notifier:
    build: ./notification_worker
    command: python worker.py
    environment:
      RABBIT_URL: amqp://guest:guest@rabbitmq:5672/
    depends_on: [rabbitmq]

# Bring the whole stack up
$ docker compose up --build
 
# In another terminal — smoke test end to end
$ curl -X POST http://localhost:8000/user/register \
    -H "Content-Type: application/json" \
    -d '{"email":"a@b.com","password":"secret","full_name":"Aditi"}'
 
$ TOKEN=$(curl -s -X POST http://localhost:8000/user/login \
    -H "Content-Type: application/json" \
    -d '{"email":"a@b.com","password":"secret"}' | jq -r .access_token)
 
$ curl -X POST http://localhost:8000/order/orders \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"user_id":1,"product_id":1,"quantity":2}'

Request Flow — What Happens When a User Places an Order

Sync vs Async — When to Use Which

Health Checks, Resilience & Failure Handling

# Order service with retries + timeout
from tenacity import retry, stop_after_attempt, wait_exponential
import httpx
 
@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=0.3, min=0.3, max=2))
def reserve_stock(product_id: int, qty: int) -> dict:
    r = httpx.post(f"{PRODUCT_URL}/reserve",
                  json={"product_id": product_id, "quantity": qty},
                  timeout=3.0)
    if r.status_code >= 500:
        r.raise_for_status()  # let tenacity retry only on 5xx
    return r.json()

Testing Microservices

Each service is a standalone FastAPI app — you can and should unit-test it in isolation using TestClient, mocking any external calls.

# tests/test_order.py
from fastapi.testclient import TestClient
from unittest.mock import patch
from order_service.main import app
 
client = TestClient(app)
 
def test_order_created_when_stock_available():
    fake_response = MagicMock(status_code=200)
    fake_response.json.return_value = {"unit_price": 1000,
                                      "remaining_stock": 10,
                                      "reserved": 1}
    with patch("order_service.main.httpx.post",
               return_value=fake_response) as mock_post, \
         patch("order_service.main.publish") as mock_pub:
        r = client.post("/orders",
                        json={"user_id":1,"product_id":1,"quantity":1})
 
    assert r.status_code == 200
    assert r.json()["total"] == 1000
    mock_post.assert_called_once()
    mock_pub.assert_called_once()      # event was published
 
def test_order_fails_on_insufficient_stock():
    fake_response = MagicMock(status_code=409)
    fake_response.json.return_value = {"detail": "Insufficient stock"}
    fake_response.raise_for_status.side_effect = httpx.HTTPStatusError(
        "409", request=None, response=fake_response)
 
    with patch("order_service.main.httpx.post", return_value=fake_response):
        r = client.post("/orders",
                        json={"user_id":1,"product_id":1,"quantity":999})
    assert r.status_code == 409

When to Use Microservices — and When to Run Away

Golden Rules