Python Flask Microservices in Practice: The Code and Commands That Really Matter

+----------------------------------------+
|        Monolithic Application         |
|  +-----+  +-------+  +-------+          |
|  | UI  |  | Auth  |  | Order |  ...     |
|  +-----+  +-------+  +-------+          |
|  +----------------------------------+   |
|  |           Database               |   |
|  +----------------------------------+   |
+----------------------------------------+

# 1 codebase, 1 deployment, 1 DB
# All functions in the same process

+---------------+  +---------------+  +---------------+
| Auth Service  |  | Order Service |  | Email Service |
| +DB users     |  | +DB orders    |  | +SMTP         |
+---------------+  +---------------+  +---------------+
        |                 |                  |
        +--------+--------+------------------+
                 |
        +--------+--------+
        | Message Broker  |
        | (Kafka/RabbitMQ)|
        +-----------------+

# N services, N deployments, N DBs (1 per service)
# Communication via HTTP/gRPC/events

# Structure of a modular monolith
myapp/
  modules/
    auth/
      api/
      services/
      models/
      tests/
    orders/
      api/
      services/
      models/
      tests/
    notifications/
  shared/
    db/
    events/        # internal bus
    logging/

# 1 deployment, but clear boundaries
# Communication via events (in-process)
# Facilitates future extraction into microservices

+-----------------+
|   API Gateway   |    <-- single entry point
+-----------------+
        |
+-------+-------+-------+
|       |       |       |
v       v       v       v
+----+ +-----+ +-----+ +------+
|Auth| |User | |Cart | |Order |    <-- microservices
+----+ +-----+ +-----+ +------+
   |       |      |       |
   v       v      v       v
+----+ +-----+ +-----+ +------+
|DB1 | |DB2  | |Redis| |DB3   |    <-- each its own DB
+----+ +-----+ +-----+ +------+
                        |
                        v
                 +-------------+
                 | Kafka/MQ    |    <-- async events
                 +-------------+

# Progressively migrate monolith -> microservices

# Step 0: Monolith
#   client -> monolith

# Step 1: Extract User Service
#   client -> API Gateway
#                |-- /users/* -> user-service (new)
#                |-- /*       -> monolith (remaining)

# Step 2: Extract Order Service
#   client -> API Gateway
#                |-- /users/*  -> user-service
#                |-- /orders/* -> order-service
#                |-- /*        -> monolith

# ... and so on until the monolith is fully dismantled

# Flask
from flask import Flask
app = Flask(__name__)

@app.route("/hello")
def hello():
    return {"message": "Hello"}

# 5 lines, 1 file, runs with: flask run

# FastAPI (for comparison)
from fastapi import FastAPI
app = FastAPI()

@app.get("/hello")
async def hello():
    return {"message": "Hello"}

# uvicorn main:app

# Django REST Framework (for comparison)
# Requires: settings.py, urls.py, views.py, manage.py...
# Much more boilerplate but A LOT of features

python -m venv .venv
source .venv/bin/activate              # Linux/Mac
.venv\Scripts\activate                 # Windows

pip install flask

# Or with classic microservice dependencies
pip install flask gunicorn requests sqlalchemy alembic \
            marshmallow flask-marshmallow flask-smorest \
            python-dotenv pytest

# Run in dev
flask --app app run --debug --port 5000

# Production microservice stack
flask                       # framework
flask-smorest               # API + auto OpenAPI
marshmallow                 # serialization (or Pydantic)
sqlalchemy[asyncio]         # ORM
alembic                     # migrations
gunicorn                    # WSGI prod
gevent                      # async workers
httpx                       # modern HTTP client
pydantic-settings           # 12-factor config
prometheus-flask-exporter   # metrics
sentry-sdk[flask]           # errors
opentelemetry-instrumentation-flask  # traces
pytest pytest-cov           # tests

# Flask 3.0+ supports async views
@app.route("/async-data")
async def get_data():
    data = await fetch_from_api()
    return data

# BUT remains WSGI: each async view blocks a worker thread
# For real async: use ASGI (FastAPI, Quart) or Gunicorn+gevent

gunicorn --worker-class gevent --workers 4 app:app
# gevent makes I/O non-blocking even with sync views

# services/order/tests/test_order_service.py
import pytest
from unittest.mock import Mock

def test_place_order_creates_order_and_outbox(db, app):
    cart_client = Mock()
    cart_client.get.return_value = {"items": [{"product_id": "p1", "qty": 2}]}
    
    catalog_client = Mock()
    catalog_client.get_product.return_value = {
        "id": "p1", "name": "Book", "price_cents": 1500,
    }
    
    service = OrderService(db, cart_client, catalog_client)
    order = service.place_order(user_id="user-1")
    
    assert order.total_cents == 3000
    assert len(order.items) == 1
    
    # Check outbox
    events = Outbox.query.all()
    assert len(events) == 1
    assert events[0].event_type == "order.created"
    assert events[0].payload["total_cents"] == 3000
    cart_client.clear.assert_called_once_with("user-1")

def test_empty_cart_raises():
    cart_client = Mock()
    cart_client.get.return_value = {"items": []}
    
    service = OrderService(db, cart_client, Mock())
    with pytest.raises(ValidationError):
        service.place_order("user-1")

# tests/integration/test_e2e_checkout.py
import pytest, httpx, time

# conftest.py starts docker-compose
@pytest.fixture(scope="session")
def stack():
    subprocess.run(["docker", "compose", "up", "-d", "--wait"], check=True)
    yield
    subprocess.run(["docker", "compose", "down", "-v"])

def test_full_checkout_flow(stack):
    base = "http://localhost"
    
    # 1. Signup
    r = httpx.post(f"{base}/api/auth/signup", json={
        "email": "alice@x.com", "password": "pass", "name": "Alice",
    })
    assert r.status_code == 201
    token = r.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}
    
    # 2. List products
    r = httpx.get(f"{base}/api/products")
    product = r.json()["items"][0]
    
    # 3. Add to cart
    r = httpx.post(
        f"{base}/api/cart/items",
        json={"product_id": product["id"], "qty": 2},
        headers=headers,
    )
    assert r.status_code == 200
    
    # 4. Checkout
    r = httpx.post(f"{base}/api/checkout", headers=headers)
    assert r.status_code == 201
    order_id = r.json()["order_id"]
    
    # 5. Wait for async processing (saga)
    for _ in range(30):        # max 30s
        r = httpx.get(f"{base}/api/orders/{order_id}", headers=headers)
        if r.json()["status"] == "paid":
            break
        time.sleep(1)
    else:
        pytest.fail("Order never reached 'paid' status")

# Verify that the REST contract between consumer (web-gateway) and provider (catalog) holds

from pact import Consumer, Provider

pact = Consumer("web-gateway").has_pact_with(Provider("catalog"), pact_dir="./pacts")

def test_get_product_contract():
    expected = {
        "id": "p1", "name": "Book", "price_cents": 1500, "stock": 10,
    }
    pact.given("product p1 exists") \
        .upon_receiving("a request for product p1") \
        .with_request("GET", "/products/p1") \
        .will_respond_with(200, body=expected)
    
    with pact:
        result = catalog_client.get_product("p1")
        assert result == expected

# Generated Pact file -> uploaded to Pact Broker
# Provider replays the pact -> verifies its API

# services/*/Dockerfile
FROM python:3.12-slim AS deps
WORKDIR /app
RUN pip install --no-cache-dir uv
COPY requirements.txt .
RUN uv pip install --system --no-cache -r requirements.txt

FROM python:3.12-slim
RUN useradd -m -u 1000 app
USER app
WORKDIR /app

COPY --from=deps /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=deps /usr/local/bin/gunicorn /usr/local/bin/gunicorn
COPY src/ ./src/

ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
EXPOSE 5000

HEALTHCHECK --interval=10s CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:5000/health')"

CMD ["gunicorn", "-c", "gunicorn.conf.py", "src.wsgi:app"]

# .github/workflows/ci.yml
name: CI
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        service: [auth, catalog, cart, order, payment, notification]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12"}
      
      - name: Install deps
        working-directory: services/${{ matrix.service }}
        run: |
          pip install -e ../../shared
          pip install -r requirements.txt
          pip install pytest pytest-cov ruff mypy
      
      - name: Lint
        working-directory: services/${{ matrix.service }}
        run: |
          ruff check src
          mypy src
      
      - name: Tests
        working-directory: services/${{ matrix.service }}
        run: pytest --cov=src --cov-report=xml
      
      - uses: codecov/codecov-action@v4
  
  integration:
    runs-on: ubuntu-latest
    needs: test
    steps:
      - uses: actions/checkout@v4
      - name: Start stack
        run: docker compose up -d --wait
      - name: Run integration tests
        run: pytest tests/integration/
      - name: Logs on failure
        if: failure()
        run: docker compose logs

# .github/workflows/cd.yml
name: CD
on:
  push:
    tags: ["v*"]

jobs:
  build:
    strategy:
      matrix:
        service: [auth, catalog, cart, order, payment, notification]
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      
      - name: Build & push
        uses: docker/build-push-action@v5
        with:
          context: services/${{ matrix.service }}
          push: true
          tags: |
            ghcr.io/myorg/${{ matrix.service }}:${{ github.ref_name }}
            ghcr.io/myorg/${{ matrix.service }}:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max
  
  deploy:
    needs: build
    environment: production
    steps:
      - uses: actions/checkout@v4
      - uses: azure/setup-helm@v4
      - name: Deploy via Helm
        run: |
          helm upgrade --install ecommerce ./k8s/charts/umbrella \
            -f ./k8s/values-prod.yaml \
            --set global.imageTag=${{ github.ref_name }} \
            --namespace prod \
            --atomic --timeout 10m