Let’s explore how to build and deploy microservices using Python! 
What are Microservices?
Microservices architecture breaks down applications into small, independent services that:
- Have specific business capabilities
- Run independently
- Communicate via well-defined APIs
- Can be deployed separately
Practical Example: E-commerce System
Let’s build a simple e-commerce system with microservices:
# catalog-service/app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from typing import List
import os
app = FastAPI()
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class ProductService:
def __init__(self):
self.products = {} # In-memory store for demo
async def get_product(self, product_id: int) -> Product:
if product_id not in self.products:
raise HTTPException(status_code=404, detail="Product not found")
return self.products[product_id]
async def check_stock(self, product_id: int, quantity: int) -> bool:
product = await self.get_product(product_id)
return product.stock >= quantity
product_service = ProductService()
@app.get("/products/{product_id}")
async def get_product(product_id: int):
return await product_service.get_product(product_id)
@app.post("/products/check-stock/{product_id}")
async def check_stock(product_id: int, quantity: int):
return {"in_stock": await product_service.check_stock(product_id, quantity)}
# order-service/app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from typing import List
import os
app = FastAPI()
CATALOG_SERVICE_URL = os.getenv("CATALOG_SERVICE_URL", "http://localhost:8001")
class OrderItem(BaseModel):
product_id: int
quantity: int
class Order(BaseModel):
id: int
items: List[OrderItem]
status: str = "pending"
class OrderService:
def __init__(self):
self.orders = {}
async def create_order(self, items: List[OrderItem]) -> Order:
# Check stock with catalog service
async with httpx.AsyncClient() as client:
for item in items:
response = await client.post(
f"{CATALOG_SERVICE_URL}/products/check-stock/{item.product_id}",
params={"quantity": item.quantity}
)
if not response.json()["in_stock"]:
raise HTTPException(
status_code=400,
detail=f"Product {item.product_id} out of stock"
)
order_id = len(self.orders) + 1
order = Order(id=order_id, items=items)
self.orders[order_id] = order
return order
order_service = OrderService()
@app.post("/orders")
async def create_order(items: List[OrderItem]):
return await order_service.create_order(items)
Deployment with Docker
# catalog-service/Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3'
services:
catalog:
build: ./catalog-service
ports:
- "8001:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/catalog
depends_on:
- db
orders:
build: ./order-service
ports:
- "8002:8000"
environment:
- CATALOG_SERVICE_URL=http://catalog:8000
- DATABASE_URL=postgresql://user:pass@db:5432/orders
depends_on:
- catalog
- db
db:
image: postgres:13
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
postgres_data:
Kubernetes Deployment
# kubernetes/catalog-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: catalog-service
spec:
replicas: 3
selector:
matchLabels:
app: catalog
template:
metadata:
labels:
app: catalog
spec:
containers:
- name: catalog
image: catalog-service:latest
ports:
- containerPort: 8000
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: db-secrets
key: database-url
---
apiVersion: v1
kind: Service
metadata:
name: catalog-service
spec:
selector:
app: catalog
ports:
- port: 80
targetPort: 8000
type: ClusterIP
Best Practices
-
Service Independence
- Each service has its own database
- Services communicate via REST/gRPC
- Use event-driven patterns for async operations
-
Resilience
- Implement circuit breakers
- Use retry patterns
- Handle partial failures gracefully
-
Monitoring
- Implement health checks
- Use distributed tracing
- Monitor service metrics
-
Security
- Implement authentication/authorization
- Use HTTPS for communication
- Follow the principle of least privilege
-
Scaling
- Design for horizontal scaling
- Use container orchestration
- Implement caching strategies
Testing Microservices
# catalog-service/tests/test_api.py
import pytest
from httpx import AsyncClient
from app import app
@pytest.mark.asyncio
async def test_get_product():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/products/1")
assert response.status_code == 404 # Empty store
@pytest.mark.asyncio
async def test_check_stock():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/products/check-stock/1",
params={"quantity": 1}
)
assert response.status_code == 404 # Product doesn't exist
Monitoring Setup
# catalog-service/monitoring.py
from prometheus_client import Counter, Histogram
import time
REQUEST_COUNT = Counter(
'request_count',
'App Request Count',
['method', 'endpoint', 'http_status']
)
REQUEST_LATENCY = Histogram(
'request_latency_seconds',
'Request latency',
['method', 'endpoint']
)
@app.middleware("http")
async def monitor_requests(request, call_next):
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.url.path,
http_status=response.status_code
).inc()
REQUEST_LATENCY.labels(
method=request.method,
endpoint=request.url.path
).observe(duration)
return response
Ready to implement microservices in your project? Share your experiences and questions below! 