Microservices

Why Python for Microservices?

Python isn't the fastest language, but for microservices its strengths shine: rapid iteration, deep AI/ML ecosystem, great HTTP/async libraries, and team velocity. Stick to Python where it makes sense; reach for Go/Rust when raw throughput dominates.

This tutorial focuses on building, communicating between, and operating Python microservices with FastAPI, gRPC, and message queues.

A Minimal FastAPI Service

# user_service/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="User Service")

class User(BaseModel):
    id: int
    email: str

users = {1: User(id=1, email="[email protected]")}

@app.get("/health")
async def health():
    return {"status": "ok"}

@app.get("/users/{uid}", response_model=User)
async def get_user(uid: int):
    if uid not in users:
        raise HTTPException(404)
    return users[uid]

Every microservice needs at minimum: a health endpoint, structured logging, metrics, and a docs route. FastAPI gives you the docs for free.

Service-to-Service HTTP

Most Python services talk to each other over HTTP/JSON. Use httpx (async) for client calls — it's the modern requests.

import httpx

async def fetch_user(uid: int) -> dict:
    async with httpx.AsyncClient(base_url="http://user-service", timeout=5.0) as client:
        r = await client.get(f"/users/{uid}")
        r.raise_for_status()
        return r.json()

Retries and Circuit Breakers

Use tenacity for retries, pybreaker for circuit breakers. Each downstream call should have a timeout, retry policy, and breaker.

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))
async def fetch_with_retries(client, url):
    r = await client.get(url)
    r.raise_for_status()
    return r.json()

gRPC with Python

For low-latency, strongly-typed RPC between services, gRPC + Protocol Buffers beats JSON over HTTP.

$ pip install grpcio grpcio-tools

1. Define the service in .proto

// user.proto
syntax = "proto3";
service UserService {
    rpc GetUser (UserRequest) returns (UserReply);
}
message UserRequest { int32 id = 1; }
message UserReply  { int32 id = 1; string email = 2; }

$ python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. user.proto
# Generates user_pb2.py and user_pb2_grpc.py

2. Server

import grpc
from concurrent.futures import ThreadPoolExecutor
import user_pb2, user_pb2_grpc

class UserServicer(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, req, ctx):
        return user_pb2.UserReply(id=req.id, email="[email protected]")

server = grpc.server(ThreadPoolExecutor(max_workers=10))
user_pb2_grpc.add_UserServiceServicer_to_server(UserServicer(), server)
server.add_insecure_port("[::]:50051")
server.start()
server.wait_for_termination()

3. Client

with grpc.insecure_channel("localhost:50051") as chan:
    stub = user_pb2_grpc.UserServiceStub(chan)
    reply = stub.GetUser(user_pb2.UserRequest(id=1))
    print(reply.email)

Message Queues

For async work (emails, jobs, event streams) decouple producers and consumers with a queue.

Celery — Distributed Task Queue

# tasks.py
from celery import Celery

app = Celery("worker", broker="redis://localhost")

@app.task
def send_email(to, subject):
    # long-running work
    smtp_send(to, subject)

# In your app — fire and forget
send_email.delay("[email protected]", "Welcome")

$ celery -A tasks worker --loglevel=info

RabbitMQ via aio-pika or pika

For pub/sub and routing, RabbitMQ is the workhorse. aio-pika is the async Python client.

Kafka via confluent-kafka or aiokafka

For event streaming, append-only logs, and big data, Kafka is the standard. aiokafka brings async support.

Observability

Every production service needs three pillars:

• Logs: structured logging with structlog or logging + JSON formatter
• Metrics: prometheus-client exposing /metrics
• Traces: OpenTelemetry — opentelemetry-instrumentation-fastapi auto-traces every request

from prometheus_client import Counter, generate_latest
from fastapi import FastAPI, Response

requests = Counter("http_requests_total", "Total HTTP requests", ["path"])

@app.middleware("http")
async def count(req, call_next):
    requests.labels(path=req.url.path).inc()
    return await call_next(req)

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type="text/plain")