5 Python Patterns I Use in Every Microservice

I reach for the same five patterns every time I start a new Python microservice. They’re not exciting, and nobody gives conference talks about them. But whenever I’ve skipped one, I’ve ended up wishing I hadn’t.

I’ve built microservices at different companies: data pipelines on RabbitMQ, backend APIs, and more recently AI services on FastAPI. The list has barely changed, and I use all five today in the Nooko.ai FastAPI services at EGS. Each pattern deals with a way services fail in production.

1. Structured config with validation

Configuration is usually the first thing to break. A missing environment variable, a typo in a database URL, a port conflict. You want to catch these at startup, not when the first request comes in.

from functools import lru_cache
 
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)
 
    service_name: str = "my-service"
    environment: str = Field(default="development")
    database_url: str  # no default, so it's required
    db_pool_min: int = 2
    db_pool_max: int = 10
    rabbitmq_url: str = "amqp://guest:guest@localhost:5672/"
    shutdown_timeout: int = 15
 
    @field_validator("environment")
    @classmethod
    def validate_environment(cls, v: str) -> str:
        allowed = {"development", "staging", "production"}
        if v not in allowed:
            raise ValueError(f"environment must be one of {allowed}")
        return v
 
 
@lru_cache
def get_settings() -> Settings:
    return Settings()

If database_url is missing, the service won’t start. You find out immediately, not after the first query fails. Every module reads config through get_settings(), so there are no stray os.getenv() calls hidden around the codebase.

A misspelled environment variable can crash a service hours after a deploy. This catches it before the process even binds to a port. It’s a small thing, but over a year it probably saves more time than anything else on this list.

2. Graceful shutdown

A service receives SIGTERM on every deploy, scale-down and container restart. When that happens it needs to stop taking new work, finish what’s in progress, close its connections and exit cleanly.

import asyncio
import logging
from collections.abc import AsyncGenerator, Awaitable, Callable
from contextlib import asynccontextmanager
 
from fastapi import FastAPI
 
logger = logging.getLogger(__name__)
 
 
class GracefulShutdown:
    def __init__(self, timeout: int = 15):
        self.timeout = timeout
        self.is_shutting_down = False
        self._tasks: set[asyncio.Task] = set()
        self._cleanup_handlers: list[Callable[[], Awaitable[None]]] = []
 
    def register_cleanup(self, handler: Callable[[], Awaitable[None]]) -> None:
        self._cleanup_handlers.append(handler)
 
    def track_task(self, task: asyncio.Task) -> None:
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
 
    async def shutdown(self) -> None:
        self.is_shutting_down = True
        if self._tasks:
            await asyncio.wait(self._tasks, timeout=self.timeout)
        for handler in reversed(self._cleanup_handlers):
            try:
                await handler()
            except Exception:
                logger.exception("Cleanup handler failed")
 
 
graceful = GracefulShutdown(timeout=15)
 
 
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    yield
    await graceful.shutdown()  # runs when the server receives SIGTERM
 
 
app = FastAPI(lifespan=lifespan)

This mattered most on the RabbitMQ services I built at Rimes for delivering client data files. Queues moved work through the pipeline step by step, and once every step was complete the result was stored in MS SQL Server. Each message was acknowledged only after its step had finished. That way, if a service stopped halfway through a message, RabbitMQ redelivered the unfinished work after the restart instead of losing it. Graceful shutdown let in-progress steps finish before the process exited, and the idempotent processing in pattern 4 meant redelivered work wasn’t duplicated.

One detail that’s easy to miss is ordering. Cleanup handlers should run in reverse, so whatever you opened last gets closed first. Database connections should close last, because other handlers may still need them during their own cleanup.

3. Health checks that actually check health

A lot of health endpoints just return {"status": "ok"}. That tells the orchestrator the process is running, not whether it can do useful work. A real health check tests each dependency:

import time
from dataclasses import dataclass, field
 
 
@dataclass
class HealthRegistry:
    checks: list[dict] = field(default_factory=list)
 
    def register(self, name: str, check, critical: bool = True) -> None:
        self.checks.append({"name": name, "check": check, "critical": critical})
 
    async def run_all(self) -> dict:
        results = {}
        all_healthy = True
        for hc in self.checks:
            start = time.monotonic()
            try:
                await hc["check"]()
                results[hc["name"]] = {
                    "status": "healthy",
                    "latency_ms": round((time.monotonic() - start) * 1000, 2),
                }
            except Exception as e:
                results[hc["name"]] = {"status": "unhealthy", "error": str(e)}
                if hc["critical"]:
                    all_healthy = False
        return {"status": "healthy" if all_healthy else "unhealthy", "checks": results}

I’d expose three endpoints. /health/live answers whether the process is alive, and an orchestrator like Kubernetes restarts the container if it fails. /health/ready answers whether it can take traffic, and the instance is pulled from the load balancer if not. /health gives full diagnostics for dashboards.

The critical flag is the important part. If Redis is down but only used for caching, the service still works, just more slowly. If the database is down, it can’t do anything. Without that distinction, a single failing non-critical check can make the orchestrator restart every instance, healthy ones included, and a minor problem becomes an outage.

4. Idempotent message processing

If you consume messages from a queue, whether that’s RabbitMQ, Kafka or SQS, you will eventually process duplicates. Network issues, consumer restarts and rebalances all cause them. Your database won’t care that it was an accident.

Here’s a simple version that uses Redis as the store. It’s only an example; any shared store works, including a table in the database you already have.

import hashlib
import json
 
 
class IdempotencyStore:
    def __init__(self, redis_client, ttl_hours: int = 24):
        self.redis = redis_client
        self.ttl_seconds = ttl_hours * 3600
 
    async def is_processed(self, message_id: str) -> bool:
        return bool(await self.redis.exists(f"idempotency:{message_id}"))
 
    async def mark_processed(self, message_id: str, result: dict | None = None) -> None:
        await self.redis.setex(
            f"idempotency:{message_id}",
            self.ttl_seconds,
            json.dumps({"result": result}),
        )
 
 
async def handle_message(message: dict, store: IdempotencyStore) -> None:
    message_id = message.get("id") or hashlib.sha256(
        json.dumps(message, sort_keys=True).encode()
    ).hexdigest()
 
    if await store.is_processed(message_id):
        return  # already handled
 
    # If this raises, the message is not marked as processed,
    # so the broker can redeliver it and we retry.
    result = await process_business_logic(message)
    await store.mark_processed(message_id, result)

The error path matters. If processing fails, the message is deliberately not marked as processed, so the broker redelivers it for a retry. Only a success gets recorded.

This is the same idea I relied on for the file delivery services at Rimes. Because a message was only acknowledged once its step was done, a restart meant RabbitMQ redelivered anything unfinished. That’s exactly when duplicates appear, so the steps had to be safe to run again without the result being stored twice in SQL Server.

If you use a TTL, as in the example, it deserves some thought. Twenty-four hours suits most cases. Too short and you miss slow retries. Too long and you pay for memory you don’t need.

5. Structured logging with request context

print() stops being useful quickly. Once you run more than a few instances, debugging a production issue means searching logs from many pods and trying to line up timestamps. You might find the failed request, but not what happened before it or who triggered it. You need searchable logs that carry context automatically.

import uuid
 
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
 
 
def setup_logging(environment: str = "development") -> None:
    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
    ]
    if environment == "production":
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())
 
    structlog.configure(processors=processors, logger_factory=structlog.PrintLoggerFactory())
 
 
class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        req_id = request.headers.get("x-request-id") or str(uuid.uuid4())
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            request_id=req_id, method=request.method, path=request.url.path
        )
        response = await call_next(request)
        response.headers["x-request-id"] = req_id
        return response

contextvars is what makes this work. You bind the context once in the middleware, and it appears in every log line for that request, in every function, without passing a logger through every method signature. When something breaks, you search by request_id and see the whole sequence.

In production every line is JSON that goes into your log aggregator. In development you get readable console output. It’s the same code with a different renderer. You set it up once and mostly forget about it until you need to debug something urgent, and then you’re glad it’s there.

Why these five

None of this is new. After nine years of professional software engineering, much of it on backend services, these are the patterns I keep coming back to. What they have in common is that they handle the unhappy path. Building a service that works when everything goes right is the easy part.

I’ve tried more elaborate approaches over the years, and most aren’t worth the complexity for a typical service. These five are simple enough that there’s little reason to skip them, and they cover the failures that cause real incidents.