Микросервисы на Python: гид по архитектуре и разработке приложений

# Создание виртуального окружения
python -m venv venv
source venv/bin/activate # Для Windows: venv\Scripts\activate

# Установка зависимостей
pip install fastapi uvicorn sqlalchemy pydantic alembic

# Создание структуры проекта
mkdir -p app/{api,models,schemas,services,database}
touch app/__init__.py app/main.py

# app/database/base.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/db_name"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

# app/models/product.py
from sqlalchemy import Column, Integer, String, Float
from app.database.base import Base

class Product(Base):
__tablename__ = "products"

id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String)
price = Column(Float)
inventory = Column(Integer)

# app/schemas/product.py
from pydantic import BaseModel

class ProductBase(BaseModel):
name: str
description: str
price: float
inventory: int

class ProductCreate(ProductBase):
pass

class Product(ProductBase):
id: int

class Config:
orm_mode = True

# app/services/product.py
from sqlalchemy.orm import Session
from app.models.product import Product
from app.schemas.product import ProductCreate

def get_product(db: Session, product_id: int):
return db.query(Product).filter(Product.id == product_id).first()

def get_products(db: Session, skip: int = 0, limit: int = 100):
return db.query(Product).offset(skip).limit(limit).all()

def create_product(db: Session, product: ProductCreate):
db_product = Product(**product.dict())
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product

# app/api/product.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from app.database.base import get_db
from app.services import product as product_service
from app.schemas.product import Product, ProductCreate

router = APIRouter()

@router.post("/products/", response_model=Product)
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
return product_service.create_product(db=db, product=product)

@router.get("/products/", response_model=List[Product])
def read_products(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
products = product_service.get_products(db, skip=skip, limit=limit)
return products

@router.get("/products/{product_id}", response_model=Product)
def read_product(product_id: int, db: Session = Depends(get_db)):
db_product = product_service.get_product(db, product_id=product_id)
if db_product is None:
raise HTTPException(status_code=404, detail="Product not found")
return db_product

# app/main.py
from fastapi import FastAPI
from app.api import product
from app.database.base import Base, engine

# Создание таблиц в базе данных
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Product Microservice")

app.include_router(product.router, tags=["products"])

@app.get("/health")
def health_check():
return {"status": "healthy"}

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

# Установка зависимостей
pip install flask flask-restful flask-sqlalchemy marshmallow

# Создание структуры проекта
mkdir -p app/{api,models,schemas,services}
touch app/__init__.py app/main.py

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/db_name'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db.init_app(app)

from app.api import product_api
app.register_blueprint(product_api)

return app

# app/models/product.py
from app import db

class Product(db.Model):
__tablename__ = 'products'

id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), index=True)
description = db.Column(db.String(500))
price = db.Column(db.Float)
inventory = db.Column(db.Integer)

# app/schemas/product.py
from marshmallow import Schema, fields

class ProductSchema(Schema):
id = fields.Int(dump_only=True)
name = fields.Str(required=True)
description = fields.Str()
price = fields.Float(required=True)
inventory = fields.Int(required=True)

# app/services/product.py
from app import db
from app.models.product import Product

def get_product(product_id):
return Product.query.get(product_id)

def get_products(skip=0, limit=100):
return Product.query.offset(skip).limit(limit).all()

def create_product(data):
product = Product(**data)
db.session.add(product)
db.session.commit()
return product

# app/api/product_api.py
from flask import Blueprint, request, jsonify
from app.services import product as product_service
from app.schemas.product import ProductSchema

product_api = Blueprint('product_api', __name__)
product_schema = ProductSchema()
products_schema = ProductSchema(many=True)

@product_api.route('/products', methods=['POST'])
def create_product():
data = request.get_json()
product = product_service.create_product(data)
return product_schema.dump(product), 201

@product_api.route('/products', methods=['GET'])
def get_products():
skip = int(request.args.get('skip', 0))
limit = int(request.args.get('limit', 100))
products = product_service.get_products(skip, limit)
return products_schema.dump(products)

@product_api.route('/products/<int:product_id>', methods=['GET'])
def get_product(product_id):
product = product_service.get_product(product_id)
if not product:
return {"message": "Product not found"}, 404
return product_schema.dump(product)

@product_api.route('/health', methods=['GET'])
def health_check():
return {"status": "healthy"}

# app/main.py
from app import create_app, db

app = create_app()

@app.before_first_request
def create_tables():
db.create_all()

if __name__ == '__main__':
app.run(host='0.0.0.0', port=8000, debug=True)

# Пример клиентского кода для взаимодействия с другим микросервисом
import requests

class ProductServiceClient:
def __init__(self, base_url="http://product-service:8000"):
self.base_url = base_url

def get_product(self, product_id):
response = requests.get(f"{self.base_url}/products/{product_id}")
if response.status_code == 404:
return None
response.raise_for_status()
return response.json()

def create_product(self, product_data):
response = requests.post(
f"{self.base_url}/products/", 
json=product_data
)
response.raise_for_status()
return response.json()

# Асинхронная версия с httpx
import httpx
import asyncio

class AsyncProductServiceClient:
def __init__(self, base_url="http://product-service:8000"):
self.base_url = base_url
self.client = httpx.AsyncClient()

async def get_product(self, product_id):
response = await self.client.get(f"{self.base_url}/products/{product_id}")
if response.status_code == 404:
return None
response.raise_for_status()
return response.json()

async def create_product(self, product_data):
response = await self.client.post(
f"{self.base_url}/products/", 
json=product_data
)
response.raise_for_status()
return response.json()

async def close(self):
await self.client.aclose()

# Отправитель сообщений (Producer)
import pika
import json

class OrderCreatedPublisher:
def __init__(self, rabbitmq_host="rabbitmq"):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host=rabbitmq_host)
)
self.channel = self.connection.channel()
self.channel.exchange_declare(
exchange='order_events', 
exchange_type='topic'
)

def publish(self, order_data):
message = json.dumps(order_data)
self.channel.basic_publish(
exchange='order_events',
routing_key='order.created',
body=message
)
print(f"Sent message: {message}")

def close(self):
self.connection.close()

# Получатель сообщений (Consumer)
import pika
import json
import threading

class OrderCreatedConsumer:
def __init__(self, callback, rabbitmq_host="rabbitmq"):
self.callback = callback
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host=rabbitmq_host)
)
self.channel = self.connection.channel()
self.channel.exchange_declare(
exchange='order_events', 
exchange_type='topic'
)
result = self.channel.queue_declare('', exclusive=True)
self.queue_name = result.method.queue
self.channel.queue_bind(
exchange='order_events',
queue=self.queue_name,
routing_key='order.created'
)

def start_consuming(self):
self.channel.basic_consume(
queue=self.queue_name,
on_message_callback=self._process_message,
auto_ack=True
)
print("Starting to consume messages...")
self.channel.start_consuming()

def _process_message(self, ch, method, properties, body):
try:
order_data = json.loads(body)
self.callback(order_data)
except Exception as e:
print(f"Error processing message: {e}")

def start(self):
thread = threading.Thread(target=self.start_consuming)
thread.daemon = True
thread.start()
return thread

def close(self):
self.channel.close()
self.connection.close()

// product.proto
syntax = "proto3";

package product;

service ProductService {
rpc GetProduct(GetProductRequest) returns (Product);
rpc CreateProduct(CreateProductRequest) returns (Product);
}

message GetProductRequest {
int32 product_id = 1;
}

message CreateProductRequest {
string name = 1;
string description = 2;
float price = 3;
int32 inventory = 4;
}

message Product {
int32 id = 1;
string name = 2;
string description = 3;
float price = 4;
int32 inventory = 5;
}

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. product.proto

# grpc_server.py
import grpc
from concurrent import futures
import product_pb2
import product_pb2_grpc
from app.services import product as product_service
from app import db, create_app

app = create_app()
app.app_context().push()

class ProductServicer(product_pb2_grpc.ProductServiceServicer):
def GetProduct(self, request, context):
product = product_service.get_product(request.product_id)
if not product:
context.set_code(grpc.StatusCode.NOT_FOUND)
context.set_details("Product not found")
return product_pb2.Product()

return product_pb2.Product(
id=product.id,
name=product.name,
description=product.description,
price=product.price,
inventory=product.inventory
)

def CreateProduct(self, request, context):
product_data = {
'name': request.name,
'description': request.description,
'price': request.price,
'inventory': request.inventory
}
product = product_service.create_product(product_data)

return product_pb2.Product(
id=product.id,
name=product.name,
description=product.description,
price=product.price,
inventory=product.inventory
)

def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
product_pb2_grpc.add_ProductServiceServicer_to_server(
ProductServicer(), server
)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

if __name__ == '__main__':
serve()

# grpc_client.py
import grpc
import product_pb2
import product_pb2_grpc

class ProductServiceClient:
def __init__(self, host="localhost:50051"):
self.channel = grpc.insecure_channel(host)
self.stub = product_pb2_grpc.ProductServiceStub(self.channel)

def get_product(self, product_id):
try:
response = self.stub.GetProduct(
product_pb2.GetProductRequest(product_id=product_id)
)
return {
'id': response.id,
'name': response.name,
'description': response.description,
'price': response.price,
'inventory': response.inventory
}
except grpc.RpcError as e:
if e.code() == grpc.StatusCode.NOT_FOUND:
return None
raise

def create_product(self, product_data):
response = self.stub.CreateProduct(
product_pb2.CreateProductRequest(
name=product_data['name'],
description=product_data.get('description', ''),
price=product_data['price'],
inventory=product_data['inventory']
)
)
return {
'id': response.id,
'name': response.name,
'description': response.description,
'price': response.price,
'inventory': response.inventory
}

def close(self):
self.channel.close()

# Dockerfile для FastAPI-микросервиса
FROM python:3.9-slim

WORKDIR /app

# Установка зависимостей
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копирование кода приложения
COPY ./app /app/app

# Открытие порта
EXPOSE 8000

# Запуск приложения с использованием Gunicorn и Uvicorn
CMD ["gunicorn", "app.main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

fastapi>=0.68.0
pydantic>=1.8.0
sqlalchemy>=1.4.0
alembic>=1.6.0
uvicorn>=0.15.0
gunicorn>=20.1.0
psycopg2-binary>=2.9.0
httpx>=0.20.0

# docker-compose.yml
version: '3'

services:
product-service:
build: 
context: ./product-service
dockerfile: Dockerfile
ports:
- "8001:8000"
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/product_db
depends_on:
- postgres
restart: on-failure

order-service:
build:
context: ./order-service
dockerfile: Dockerfile
ports:
- "8002:8000"
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/order_db
- PRODUCT_SERVICE_URL=http://product-service:8000
depends_on:
- postgres
- product-service
restart: on-failure

postgres:
image: postgres:13
ports:
- "5432:5432"
environment:
- POSTGRES_PASSWORD=password
- POSTGRES_USER=postgres
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql

rabbitmq:
image: rabbitmq:3-management
ports:
- "5672:5672"
- "15672:15672"

volumes:
postgres_data:

# kubernetes/product-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: product-service
spec:
replicas: 3
selector:
matchLabels:
app: product-service
template:
metadata:
labels:
app: product-service
spec:
containers:
- name: product-service
image: your-registry/product-service:latest
ports:
- containerPort: 8000
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: db-secrets
key: database-url
resources:
requests:
memory: "128Mi"
cpu: "100m"
limits:
memory: "256Mi"
cpu: "200m"
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 15
periodSeconds: 20

# kubernetes/product-service.yaml
apiVersion: v1
kind: Service
metadata:
name: product-service
spec:
selector:
app: product-service
ports:
- port: 80
targetPort: 8000
type: ClusterIP

# kubernetes/db-secrets.yaml
apiVersion: v1
kind: Secret
metadata:
name: db-secrets
type: Opaque
data:
# Закодированный в base64 строки подключения
database-url: cG9zdGdyZXNxbDovL3Bvc3RncmVzOnBhc3N3b3JkQHBvc3RncmVzOjU0MzIvcHJvZHVjdF9kYg==

# app/monitoring.py
from prometheus_client import Counter, Histogram, generate_latest
import time

REQUEST_COUNT = Counter(
'http_requests_total', 
'Total HTTP Requests',
['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
'http_request_duration_seconds',
'HTTP request latency in seconds',
['method', 'endpoint']
)

def start_request_timer():
return time.time()

def end_request_timer(method, endpoint, start_time):
latency = time.time() – start_time
REQUEST_LATENCY.labels(method=method, endpoint=endpoint).observe(latency)

def increment_request_count(method, endpoint, status):
REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status).inc()

# app/main.py
from fastapi import FastAPI, Request, Response
from fastapi.responses import PlainTextResponse
from app.monitoring import start_request_timer, end_request_timer, increment_request_count
from prometheus_client import generate_latest

app = FastAPI(title="Product Microservice")

@app.middleware("http")
async def monitor_requests(request: Request, call_next):
method = request.method
endpoint = request.url.path

start_time = start_request_timer()
response = await call_next(request)
end_request_timer(method, endpoint, start_time)

status_code = response.status_code
increment_request_count(method, endpoint, status_code)

return response

@app.get("/metrics", response_class=PlainTextResponse)
async def metrics():
return generate_latest()

# Остальной код приложения...

# .github/workflows/ci-cd.yml
name: Python Microservice CI/CD

on:
push:
branches: [ main ]
pull_request:
branches: [ main ]

jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-cov
- name: Test with pytest
run: |
pytest --cov=app tests/

build:
needs: test
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v2
- name: Build and push Docker image
uses: docker/build-push-action@v2
with:
context: .
push: true
tags: your-registry/product-service:latest

deploy:
needs: build
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v2
- name: Deploy to Kubernetes
uses: actions-hub/kubectl@master
env:
KUBE_CONFIG: ${{ secrets.KUBE_CONFIG }}
with:
args: apply -f kubernetes/