9 Essential Components of a Production Microservice App¶

# Registering with Consul
import consul

c = consul.Consul(host="consul-server", port=8500)

# Register this service instance
c.agent.service.register(
    name="order-service",
    service_id="order-service-instance-1",
    address="10.0.0.5",
    port=8002,
    tags=["v2", "production"],
    check=consul.Check.http(
        "http://10.0.0.5:8002/health",
        interval="10s",
        timeout="5s",
        deregister="30s"  # auto-deregister after 30s of failure
    )
)

# Discover a service
_, services = c.health.service("order-service", passing=True)
instances = [(s["Service"]["Address"], s["Service"]["Port"]) for s in services]
print(instances)  # [("10.0.0.5", 8002), ("10.0.0.6", 8002)]

# Each microservice is a small, focused FastAPI application
# order_service/main.py

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from decimal import Decimal
from datetime import datetime
import httpx

app = FastAPI(title="Order Service", version="2.1.0")

class OrderCreate(BaseModel):
    user_id: str
    product_id: str
    quantity: int
    shipping_address: str

class OrderResponse(BaseModel):
    order_id: str
    status: str
    total_amount: Decimal
    created_at: datetime

@app.get("/health")
def health_check():
    """Health check for readiness probe."""
    return {"status": "healthy", "service": "order-service", "version": "2.1.0"}

@app.post("/orders", response_model=OrderResponse)
async def create_order(order: OrderCreate):
    # 1. Validate product exists (call Product Service)
    async with httpx.AsyncClient() as client:
        product_resp = await client.get(f"http://product-service/products/{order.product_id}")
        if product_resp.status_code == 404:
            raise HTTPException(404, "Product not found")
        product = product_resp.json()

    # 2. Calculate total
    total = Decimal(str(product["price"])) * order.quantity

    # 3. Save order to this service's database
    order_id = await save_order_to_db(order, total)

    # 4. Publish event for other services to react
    await publish_event("order.created", {
        "order_id": order_id,
        "user_id": order.user_id,
        "total": str(total)
    })

    return OrderResponse(
        order_id=order_id,
        status="confirmed",
        total_amount=total,
        created_at=datetime.utcnow()
    )

# Prevent cascading failures when a dependency is down
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_product_service(product_id: str) -> dict:
    """If this fails 5 times, circuit opens for 30 seconds."""
    async with httpx.AsyncClient(timeout=3.0) as client:
        response = await client.get(f"http://product-service/products/{product_id}")
        response.raise_for_status()
        return response.json()

async def get_product_safe(product_id: str) -> dict:
    try:
        return await call_product_service(product_id)
    except Exception:
        # Fallback: return cached or default data
        return {"product_id": product_id, "price": 0, "name": "Unknown", "available": False}

# Authorization Server using python-jose for JWT
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

app = FastAPI(title="Authorization Server")

SECRET_KEY = "your-256-bit-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

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

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str
    expires_in: int

def create_token(data: dict, expires_delta: timedelta) -> str:
    payload = {**data, "exp": datetime.utcnow() + expires_delta, "iat": datetime.utcnow()}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/auth/token", response_model=Token)
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = await get_user_from_db(form.username)
    if not user or not pwd_context.verify(form.password, user.hashed_password):
        raise HTTPException(401, "Invalid credentials", headers={"WWW-Authenticate": "Bearer"})

    access_token = create_token(
        data={"sub": user.id, "email": user.email, "roles": user.roles},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_token(
        data={"sub": user.id, "type": "refresh"},
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )

    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer",
        expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )

# Middleware for services to verify tokens (shared library)
def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") == "refresh":
            raise HTTPException(401, "Cannot use refresh token for API access")
        return payload
    except JWTError:
        raise HTTPException(401, "Could not validate credentials")

def require_role(role: str):
    def checker(user: dict = Depends(get_current_user)):
        if role not in user.get("roles", []):
            raise HTTPException(403, f"Role '{role}' required")
        return user
    return checker

# In any downstream service:
@app.delete("/users/{user_id}")
async def delete_user(user_id: str, admin=Depends(require_role("admin"))):
    # Only admins can reach here
    await remove_user(user_id)
    return {"deleted": user_id}

# database/connection.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import random

# Separate engines for write and read
WRITE_ENGINE = create_engine(
    "postgresql://user:pass@primary-db:5432/orders",
    pool_size=10,
    max_overflow=20,
)

READ_ENGINES = [
    create_engine(f"postgresql://user:pass@replica-{i}:5432/orders", pool_size=10)
    for i in range(1, 4)  # 3 read replicas
]

def get_write_session():
    """Use for INSERT, UPDATE, DELETE."""
    Session = sessionmaker(bind=WRITE_ENGINE)
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

def get_read_session():
    """Use for SELECT — load balanced across replicas."""
    engine = random.choice(READ_ENGINES)  # or use round-robin
    Session = sessionmaker(bind=engine)
    session = Session()
    try:
        yield session
    finally:
        session.close()

# Usage in FastAPI endpoints
from fastapi import Depends

@app.get("/orders/{order_id}")
async def get_order(order_id: str, db=Depends(get_read_session)):
    return db.query(Order).filter(Order.id == order_id).first()

@app.post("/orders")
async def create_order(order: OrderCreate, db=Depends(get_write_session)):
    new_order = Order(**order.dict())
    db.add(new_order)
    # commit happens automatically in context manager
    return new_order

# cache/redis_client.py
import redis.asyncio as redis
import json
from functools import wraps
from typing import Any, Optional, Callable
import hashlib

class CacheManager:
    def __init__(self, redis_url: str = "redis://redis-cluster:6379"):
        self.redis = redis.from_url(redis_url, decode_responses=True)

    async def get(self, key: str) -> Optional[Any]:
        value = await self.redis.get(key)
        return json.loads(value) if value else None

    async def set(self, key: str, value: Any, ttl: int = 300) -> None:
        await self.redis.setex(key, ttl, json.dumps(value, default=str))

    async def delete(self, key: str) -> None:
        await self.redis.delete(key)

    async def delete_pattern(self, pattern: str) -> None:
        """Delete all keys matching a pattern — use carefully."""
        keys = await self.redis.keys(pattern)
        if keys:
            await self.redis.delete(*keys)

cache = CacheManager()

# Decorator for easy caching
def cached(ttl: int = 300, key_prefix: str = ""):
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Build cache key from function name + arguments
            key_data = f"{key_prefix}:{func.__name__}:{args}:{sorted(kwargs.items())}"
            cache_key = hashlib.md5(key_data.encode()).hexdigest()

            # Try cache first
            cached_result = await cache.get(cache_key)
            if cached_result is not None:
                return cached_result

            # Cache miss — call the real function
            result = await func(*args, **kwargs)
            await cache.set(cache_key, result, ttl=ttl)
            return result
        return wrapper
    return decorator

# Usage
@cached(ttl=60, key_prefix="product")  # cache for 60 seconds
async def get_product(product_id: str) -> dict:
    return await db.query(f"SELECT * FROM products WHERE id = '{product_id}'")

# Cache invalidation when data changes
async def update_product(product_id: str, data: dict):
    await db.update("products", product_id, data)
    await cache.delete_pattern(f"*product*{product_id}*")  # invalidate all related cache

# Store user sessions in Redis (shared across all service instances)
from fastapi import Request, Response
import uuid

SESSION_TTL = 3600  # 1 hour

async def create_session(user_id: str) -> str:
    session_id = str(uuid.uuid4())
    await cache.set(
        f"session:{session_id}",
        {"user_id": user_id, "created_at": str(datetime.utcnow())},
        ttl=SESSION_TTL
    )
    return session_id

async def get_session(session_id: str) -> Optional[dict]:
    session = await cache.get(f"session:{session_id}")
    if session:
        # Extend TTL on activity (sliding window)
        await cache.redis.expire(f"session:{session_id}", SESSION_TTL)
    return session

async def delete_session(session_id: str) -> None:
    await cache.delete(f"session:{session_id}")

# messaging/publisher.py
import aio_pika
import json
from datetime import datetime

async def publish_event(event_type: str, payload: dict) -> None:
    """Publish an event to the message broker."""
    connection = await aio_pika.connect_robust("amqp://user:pass@rabbitmq/")

    async with connection:
        channel = await connection.channel()

        # Declare exchange (creates if doesn't exist)
        exchange = await channel.declare_exchange(
            "domain_events",
            aio_pika.ExchangeType.TOPIC,
            durable=True  # survive broker restart
        )

        message_body = {
            "event_type": event_type,
            "payload": payload,
            "timestamp": datetime.utcnow().isoformat(),
            "source": "order-service"
        }

        await exchange.publish(
            aio_pika.Message(
                body=json.dumps(message_body).encode(),
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,  # survive broker restart
                content_type="application/json"
            ),
            routing_key=event_type  # e.g., "order.created"
        )
        print(f"Published: {event_type}")

# messaging/consumer.py
async def start_consumer(queue_name: str, routing_pattern: str):
    """Subscribe to events and process them."""
    connection = await aio_pika.connect_robust("amqp://user:pass@rabbitmq/")
    channel = await connection.channel()
    await channel.set_qos(prefetch_count=10)  # process 10 messages at a time

    exchange = await channel.declare_exchange("domain_events", aio_pika.ExchangeType.TOPIC)
    queue = await channel.declare_queue(queue_name, durable=True)
    await queue.bind(exchange, routing_key=routing_pattern)

    async def process_message(message: aio_pika.abc.AbstractIncomingMessage):
        async with message.process():  # auto-ack on success, nack on exception
            event = json.loads(message.body)
            print(f"Processing: {event['event_type']}")
            await handle_event(event)

    await queue.consume(process_message)
    print(f"Consuming from {queue_name}, pattern: {routing_pattern}")

# Event handlers in different services

# Notification Service listens for order events
async def handle_event(event: dict):
    if event["event_type"] == "order.created":
        await send_confirmation_email(event["payload"]["user_id"])
    elif event["event_type"] == "order.shipped":
        await send_shipping_notification(event["payload"])

# Start the consumer (in a FastAPI lifespan or separate worker)
import asyncio
asyncio.run(start_consumer("notification-queue", "order.*"))

# kafka/producer.py — millions of events per second
from confluent_kafka import Producer
import json

producer = Producer({"bootstrap.servers": "kafka-1:9092,kafka-2:9092,kafka-3:9092"})

def delivery_callback(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} partition [{msg.partition()}]")

def publish_to_kafka(topic: str, key: str, value: dict):
    producer.produce(
        topic=topic,
        key=key.encode(),
        value=json.dumps(value).encode(),
        callback=delivery_callback
    )
    producer.poll(0)  # trigger callbacks

# kafka/consumer.py
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092,kafka-2:9092",
    "group.id": "payment-service-group",  # consumers in same group share partitions
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False   # manual commit for exactly-once processing
})

consumer.subscribe(["order.created"])

while True:
    msg = consumer.poll(1.0)  # wait up to 1 second
    if msg and not msg.error():
        event = json.loads(msg.value())
        process_order_event(event)
        consumer.commit()  # commit only after successful processing

# metrics/instrumentation.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from fastapi import FastAPI, Request, Response
import time

# Define metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status_code", "service"]
)

REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration in seconds",
    ["method", "endpoint"],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

ACTIVE_CONNECTIONS = Gauge(
    "http_active_connections",
    "Currently active HTTP connections"
)

DB_QUERY_DURATION = Histogram(
    "database_query_duration_seconds",
    "Database query duration",
    ["operation", "table"]
)

CACHE_HITS = Counter("cache_hits_total", "Cache hits", ["cache_type"])
CACHE_MISSES = Counter("cache_misses_total", "Cache misses", ["cache_type"])

# FastAPI middleware to auto-instrument all requests
app = FastAPI()

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    ACTIVE_CONNECTIONS.inc()
    start = time.time()

    response = await call_next(request)

    duration = time.time() - start
    endpoint = request.url.path

    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=endpoint,
        status_code=response.status_code,
        service="order-service"
    ).inc()

    REQUEST_DURATION.labels(
        method=request.method,
        endpoint=endpoint
    ).observe(duration)

    ACTIVE_CONNECTIONS.dec()
    return response

@app.get("/metrics")
def metrics():
    """Prometheus scrapes this endpoint."""
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

# Usage in service code:
async def get_orders_from_db(user_id: str):
    with DB_QUERY_DURATION.labels(operation="select", table="orders").time():
        return await db.execute("SELECT * FROM orders WHERE user_id = $1", user_id)

# logging/setup.py
import structlog
import logging
import sys

def configure_logging(service_name: str, version: str):
    """Configure structured JSON logging for the service."""

    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_log_level,
            structlog.stdlib.add_logger_name,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()     # output as JSON!
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

log = structlog.get_logger()

# Usage in your services
log.info("order.created",
    order_id="ORD-12345",
    user_id="USR-789",
    amount=299.99,
    currency="USD",
    trace_id="abc123",        # for cross-service correlation
    service=service_name,
    version=version
)

log.error("payment.failed",
    order_id="ORD-12345",
    error_code="card_declined",
    error_message="Insufficient funds",
    trace_id="abc123",
    exc_info=True
)

# Configure Elasticsearch for optimal log storage
from elasticsearch import Elasticsearch

es = Elasticsearch(["http://elasticsearch:9200"])

# Index template for all microservice logs
es.indices.put_index_template(
    name="microservices-logs",
    body={
        "index_patterns": ["microservices-logs-*"],
        "template": {
            "settings": {
                "number_of_shards": 3,
                "number_of_replicas": 1,
                "index.lifecycle.name": "logs-policy",  # auto-delete after 30 days
            },
            "mappings": {
                "properties": {
                    "@timestamp":    {"type": "date"},
                    "level":         {"type": "keyword"},
                    "service":       {"type": "keyword"},
                    "trace_id":      {"type": "keyword"},
                    "user_id":       {"type": "keyword"},
                    "order_id":      {"type": "keyword"},
                    "event":         {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
                    "error_message": {"type": "text"},
                }
            }
        }
    }
)

# Search logs by trace_id to follow a request across services
results = es.search(index="microservices-logs-*", body={
    "query": {
        "term": {"trace_id": "abc123"}
    },
    "sort": [{"@timestamp": "asc"}],
    "size": 100
})

for hit in results["hits"]["hits"]:
    source = hit["_source"]
    print(f"[{source['@timestamp']}] [{source['service']}] {source['event']}")