Category: python
Complete Python Microservices Project - Architecture
Published on 22 Aug 2026
Explanation
A complete Python microservices project combines the major concepts learned throughout the microservices series. In this project, an e-commerce application is divided into independent services such as User Service, Product Service, Order Service, and Payment Service. An API Gateway acts as the entry point for clients and routes requests to the appropriate service. Each service can have its own business logic and database. Docker Compose is used to run all services together. The architecture can later be extended with service discovery, load balancing, authentication, centralized configuration, logging, tracing, retries, and circuit breakers.
Code:
# Complete Architecture # # Client # | # v # API Gateway # :8000 # | # +----------------+----------------+ # | | | # v v v # User Service Product Service Order Service # :8001 :8002 :8003 # | # v # Payment Service # :8004 # # Each service can have its own database. # Docker Compose manages all containers.
Explanation
The Product Service manages product-related functionality. It exposes REST endpoints for retrieving products and individual product details. FastAPI is used to create the service, while Pydantic models provide request and response validation. In a production project, the in-memory data would be replaced with a database such as PostgreSQL or MySQL.
Code:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title='Product Service')
class Product(BaseModel):
name: str
price: float
products = {
1: {
'id': 1,
'name': 'Laptop',
'price': 65000
},
2: {
'id': 2,
'name': 'Mouse',
'price': 750
}
}
@app.get('/products')
def get_products():
return list(products.values())
@app.get('/products/{product_id}')
def get_product(product_id: int):
product = products.get(product_id)
if not product:
raise HTTPException(
status_code=404,
detail='Product not found'
)
return product
@app.get('/health')
def health():
return {'status': 'healthy'}
Explanation
The Order Service manages order-related operations and communicates with the Product Service to retrieve product information. HTTPX is used for asynchronous REST communication between services. The Product Service URL is loaded from an environment variable so that the same application can run locally, inside Docker Compose, or in another environment without changing the source code.
Code:
import os
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title='Order Service')
PRODUCT_SERVICE_URL = os.getenv(
'PRODUCT_SERVICE_URL',
'http://localhost:8002'
)
class OrderRequest(BaseModel):
product_id: int
quantity: int
@app.post('/orders')
async def create_order(order: OrderRequest):
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f'{PRODUCT_SERVICE_URL}/products/{order.product_id}',
timeout=5.0
)
if response.status_code == 404:
raise HTTPException(
status_code=404,
detail='Product not found'
)
response.raise_for_status()
product = response.json()
total = product['price'] * order.quantity
return {
'product_id': order.product_id,
'quantity': order.quantity,
'total': total,
'status': 'created'
}
except httpx.RequestError:
raise HTTPException(
status_code=503,
detail='Product Service unavailable'
)
Explanation
The API Gateway provides a single public entry point for clients. Instead of exposing every internal service directly, clients communicate with the gateway. The gateway forwards requests to User, Product, Order, and Payment Services. It can also handle authentication, authorization, rate limiting, CORS, request IDs, logging, and other cross-cutting concerns. For a learning project, FastAPI and HTTPX are enough to implement a simple gateway.
Code:
import httpx
from fastapi import FastAPI
app = FastAPI(title='API Gateway')
PRODUCT_SERVICE = 'http://product-service:8000'
ORDER_SERVICE = 'http://order-service:8000'
@app.get('/products')
async def products():
async with httpx.AsyncClient() as client:
response = await client.get(
f'{PRODUCT_SERVICE}/products',
timeout=5.0
)
return response.json()
@app.post('/orders')
async def orders(data: dict):
async with httpx.AsyncClient() as client:
response = await client.post(
f'{ORDER_SERVICE}/orders',
json=data,
timeout=5.0
)
return response.json()
@app.get('/health')
def health():
return {'status': 'healthy'}
Explanation
Docker Compose allows all Python microservices to be built and started together. Each service has its own Dockerfile and container. Docker Compose provides networking between the services, so services can communicate using names such as product-service and order-service instead of fixed IP addresses. Environment variables are used to configure service URLs. This makes the project reproducible and easy to run on different development machines.
Code:
services:
api-gateway:
build: ./api-gateway
ports:
- '8000:8000'
depends_on:
- product-service
- order-service
product-service:
build: ./product-service
ports:
- '8001:8000'
order-service:
build: ./order-service
ports:
- '8002:8000'
environment:
PRODUCT_SERVICE_URL: http://product-service:8000
depends_on:
- product-service
payment-service:
build: ./payment-service
ports:
- '8003:8000'
# Start everything:
# docker compose up --build
# Stop everything:
# docker compose down