Building a Python microservice: best practices explained

Building a Python microservice: best practices explained

By day 83 of serious backend work, you've likely felt the pain of a monolith that does everything. A single service that handles user auth, payments, notifications, and inventory reporting becomes a bottleneck—hard to deploy, hard to scale, hard to reason about. Building a Python microservice the right way means understanding that small, focused services that are easy to run aren't just an architectural preference; they're a practical necessity when teams need to move fast and deploy independently.

In this article, we'll walk through the core principles that separate a resilient microservice from a distributed mess. We'll cover how to define responsibility boundaries, manage data isolation, coordinate across services, and standardize operations. By the end, you'll have a concrete mental model for architecting Python services that your team can actually own and operate.

One responsibility — A clear boundary around one business capability

The first rule of microservices is also the hardest to enforce: each service owns exactly one business capability. Not "mostly one," not "usually one"—exactly one. When we violate this, we end up recreating the monolith in pieces.

Let's say you're building an e-commerce platform. Your first instinct might be to create a "product service" that handles product catalog, pricing, inventory, and recommendations. That's actually four responsibilities hidden in one name. Instead, think smaller: a product catalog service exposes product metadata (name, description, category). Pricing is a separate concern that might be time-based or user-based. Inventory is its own domain with its own state transitions. Recommendations could be another service entirely, pulling data from the others.

In practice, this boundary becomes your service's API contract. When you deploy a Python service, you should be able to describe its responsibility in one clear sentence:

"The order-fulfillment service accepts orders and coordinates with warehouse systems to pack and ship."

Not: "handles orders, manages returns, processes refunds, and tracks customer history."

Here's what a responsibility-focused service structure looks like:

# order_fulfillment_service/
# ├── main.py
# ├── handlers.py           # HTTP request handling
# ├── domain/
# │   ├── order.py         # Business logic for orders
# │   ├── fulfillment.py   # Fulfillment state machine
# │   └── exceptions.py
# ├── adapters/
# │   ├── warehouse_client.py   # Calls warehouse API
# │   └── event_publisher.py    # Publishes events
# ├── repository.py        # Data persistence
# └── tests/

# In handlers.py - single responsibility
@router.post("/orders/{order_id}/fulfill")
async def fulfill_order(order_id: str):
    order = repository.get(order_id)
    result = fulfillment.process(order)
    
    if result.success:
        warehouse.notify(result.warehouse_request)
        event_publisher.publish("order.fulfilled", result)
        return {"status": "fulfilling"}
    else:
        return {"error": result.error}, 400

When a service gets too many responsibilities, you can't deploy it independently, you can't scale it independently, and you can't reason about it independently. Keep the boundary tight.

Own your data — A private database, never shared tables

One of the most common mistakes we see is services that share a database. "It's faster," people say. "We can use foreign keys," they add. Then six months later, you can't deploy the payment service without coordinating with the order service, and you can't scale the inventory service without scaling everything else.

Every microservice must own its database. Not a schema, not a namespace—a completely separate database instance or at minimum a separate set of tables with exclusive ownership. This is non-negotiable.

Why? Because data is the service's internal implementation detail. When you query another service's tables directly, you're creating a hidden coupling. If the other team changes their schema, your service breaks. If they need to migrate to a different database engine, they have to coordinate with you. If they want to encrypt sensitive fields, they have to check all the places their tables are accessed.

A Python service owns its data like this:

# payment_service/repository.py
import os
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session

DATABASE_URL = os.getenv("PAYMENT_DB_URL", "postgresql://user:pass@localhost/payment_db")
engine = create_engine(DATABASE_URL)
Base = declarative_base()

class Transaction(Base):
    __tablename__ = "transactions"  # Owned by payment_service only
    
    id = Column(String, primary_key=True)
    user_id = Column(String)  # Foreign key reference, not actual FK
    amount = Column(Integer)
    status = Column(String)
    created_at = Column(String)

def save_transaction(tx: Transaction):
    with Session(engine) as session:
        session.add(tx)
        session.commit()

def get_transaction(tx_id: str) -> Transaction:
    with Session(engine) as session:
        return session.query(Transaction).filter_by(id=tx_id).first()

Notice that user_id is a reference, not a foreign key constraint to the user service's database. We store it for context, but we never join across service boundaries.

When other services need data from your service, they call your API. When you need data from another service, you call their API (and cache if necessary). This keeps data contracts explicit and prevents tight coupling.

Communicate with events — Kafka or RabbitMQ for async integration

Microservices need to talk to each other, but synchronous calls create coupling and cascade failures. If service A calls service B which calls service C, and C is slow or down, A's users see timeouts. Instead, services should communicate asynchronously through events.

Events are facts about what happened. When an order is placed, the order service publishes an "order.created" event. The payment service, shipping service, and analytics service subscribe to it and act independently. If one subscriber is temporarily slow, it doesn't block the others.

Here's a practical setup using Kafka with a Python publisher and subscriber:

# order_service/event_publisher.py
from kafka import KafkaProducer
import json
import os

broker = os.getenv("KAFKA_BROKER", "localhost:9092")
producer = KafkaProducer(
    bootstrap_servers=[broker],
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

def publish_order_event(event_type: str, order_id: str, data: dict):
    """Publish an order event to Kafka."""
    message = {
        "event_type": event_type,
        "order_id": order_id,
        "data": data,
        "timestamp": int(time.time())
    }
    producer.send("order-events", value=message)
    print(f"Published {event_type} for order {order_id}")

And a subscriber in another service:

# payment_service/event_consumer.py
from kafka import KafkaConsumer
import json
import os

broker = os.getenv("KAFKA_BROKER", "localhost:9092")
consumer = KafkaConsumer(
    "order-events",
    bootstrap_servers=[broker],
    group_id="payment_service",
    value_deserializer=lambda m: json.loads(m.decode("utf-8"))
)

def handle_orders():
    """Listen for order events and process payments."""
    for message in consumer:
        event = message.value
        if event["event_type"] == "order.created":
            order_id = event["order_id"]
            amount = event["data"]["total"]
            
            try:
                process_payment(order_id, amount)
                publish_event("payment.processed", order_id)
            except Exception as e:
                publish_event("payment.failed", order_id, {"error": str(e)})
                logger.error(f"Payment failed for {order_id}: {e}")

if __name__ == "__main__":
    handle_orders()

Event-driven architecture gives you loose coupling, independent scaling, and resilience. If the payment service is down for 10 minutes, orders keep flowing in; the payment service catches up when it recovers. That's the power of async communication.

Standard tooling — Same logging, metrics and deployment pipeline as every other service

When you have five microservices, each with different logging formats, different monitoring dashboards, and different deployment procedures, operational burden explodes. Standardize ruthlessly.

Every Python service in your fleet should have:

• Structured JSON logging to stdout (no application-specific logs to files)
• Health check endpoints
• Prometheus metrics exports
• Same Docker image build process
• Same Kubernetes manifests or deployment process

Here's a standardized setup:

# shared/logging_config.py
import json
import logging
import sys
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "service": os.getenv("SERVICE_NAME", "unknown")
        }
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_obj)

def setup_logging():
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JSONFormatter())
    logging.root.addHandler(handler)
    logging.root.setLevel(logging.INFO)

# main.py
from shared.logging_config import setup_logging
from prometheus_client import Counter, Histogram, start_http_server

setup_logging()
logger = logging.getLogger(__name__)

# Standard metrics
request_count = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)
request_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint']
)

@app.get("/health")
async def health_check():
    """Standard health check endpoint."""
    return {"status": "healthy", "service": "order-service"}

if __name__ == "__main__":
    start_http_server(8001)  # Metrics on port 8001
    uvicorn.run(app, host="0.0.0.0", port=8000)

When every service logs the same way, you can pipe all logs to a central aggregator (ELK, Loki, Datadog) and search across services instantly. When every service exports metrics the same way, your Prometheus setup works everywhere. When every service deploys the same way, your team can move between services without relearning deployment procedures.

Common mistakes

Mistake 1: Treating "microservices" as "small code." A service with 500 lines can still violate these principles. The size doesn't matter; the boundary and independence do.

Mistake 2: Sharing databases for convenience. It's faster to join across services at the database level, but you're paying that cost back tenfold in coupling and operational pain. Own your data.

Mistake 3: Overusing synchronous calls. Every sync call introduces a dependency and a failure point. Prefer events. For truly synchronous requirements (like a user needs a result immediately), use sync calls sparingly and always implement timeouts and circuit breakers.

Mistake 4: Different operational approaches per service. When the payment service deploys differently than the inventory service, your team spends time learning service-specific procedures instead of solving business problems. Standardization is investment.

Key takeaways

• Each microservice should own exactly one business capability; if you need two sentences to describe it, split it.
• Every service gets its own database; shared tables create hidden coupling that breaks independence.
• Design for async communication via Kafka or RabbitMQ where possible; sync calls should be the exception with proper timeouts and fallbacks.
• Standardize logging, metrics, health checks, and deployment across all services; operational consistency reduces cognitive load and enables faster incident response.
• Start with a clear service boundary before writing code; ambiguous responsibilities lead to scope creep and defeat the purpose of microservices.