← Back to all products
$49
Python Microservices Kit
Microservice scaffolds with gRPC, message queues, service discovery, circuit breakers, and distributed tracing.
DockerPythonMarkdownJSONYAMLFastAPIRedis
📁 File Structure 19 files
python-microservices-kit/
├── LICENSE
├── README.md
├── configs/
│ └── docker-compose.prod.yml
├── docker-compose.yml
├── guides/
│ └── microservices-patterns.md
├── services/
│ ├── api-gateway/
│ │ ├── Dockerfile
│ │ └── main.py
│ ├── order-service/
│ │ ├── Dockerfile
│ │ ├── main.py
│ │ └── models.py
│ └── user-service/
│ ├── Dockerfile
│ ├── main.py
│ └── models.py
├── shared/
│ ├── events.py
│ ├── health.py
│ ├── service_client.py
│ └── tracing.py
└── tests/
└── test_integration.py
📖 Documentation Preview README excerpt
Python Microservices Kit
Production-ready microservices architecture with API gateway, service discovery, event-driven communication, and Docker orchestration.
---
What You Get
- API Gateway with rate limiting, JWT auth, and request routing
- User Service and Order Service as example bounded contexts
- Shared library for events, service clients, health checks, and tracing
- Docker Compose configs for dev and production
- Integration tests covering cross-service workflows
- Architecture guide explaining patterns and trade-offs
File Tree
python-microservices-kit/
├── README.md
├── manifest.json
├── LICENSE
├── docker-compose.yml
├── services/
│ ├── api-gateway/
│ │ ├── main.py # Gateway with routing & rate limiting
│ │ └── Dockerfile
│ ├── user-service/
│ │ ├── main.py # User CRUD + auth endpoints
│ │ ├── models.py # User domain models
│ │ └── Dockerfile
│ └── order-service/
│ ├── main.py # Order management endpoints
│ ├── models.py # Order domain models
│ └── Dockerfile
├── shared/
│ ├── events.py # Event bus & domain events
│ ├── service_client.py # HTTP client with retries & circuit breaker
│ ├── health.py # Health check protocol
│ └── tracing.py # Distributed tracing with correlation IDs
├── configs/
│ └── docker-compose.prod.yml # Production overrides
├── tests/
│ └── test_integration.py # Cross-service integration tests
└── guides/
└── microservices-patterns.md # Architecture patterns reference
Getting Started
1. Run with Docker Compose
docker-compose up --build
Services start on:
| Service | Port |
|----------------|-------|
| API Gateway | 8000 |
... continues with setup instructions, usage examples, and more.
📄 Code Sample .py preview
shared/events.py
"""Event bus and domain events for inter-service communication.
Provides an in-process async pub/sub system that can be swapped for
RabbitMQ, Kafka, or Redis Streams in production deployments.
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable
from uuid import uuid4
logger = logging.getLogger(__name__)
# Type alias for event handlers
EventHandler = Callable[["DomainEvent"], Awaitable[None]]
@dataclass
class DomainEvent:
"""Immutable domain event transmitted between services."""
type: str
payload: dict[str, Any] = field(default_factory=dict)
event_id: str = field(default_factory=lambda: uuid4().hex)
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
source: str = ""
def to_dict(self) -> dict[str, Any]:
"""Serialise the event for transport (e.g. message queue)."""
return {
"event_id": self.event_id,
"type": self.type,
"payload": self.payload,
"timestamp": self.timestamp,
"source": self.source,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DomainEvent:
"""Reconstruct an event from a serialised dict."""
return cls(
type=data["type"],
payload=data.get("payload", {}),
event_id=data.get("event_id", uuid4().hex),
timestamp=data.get("timestamp", datetime.now(timezone.utc).isoformat()),
source=data.get("source", ""),
# ... 52 more lines ...