Микросервисы — FastAPI Pro: От новичка до

Микросервисы с FastAPI

Микросервисная архитектура для масштабируемых приложений.

#Монолит vs Микросервисы

Критерий	Монолит	Микросервисы
Размер	Одно приложение	Много небольших сервисов
Развёртывание	Одно целое	Независимое
Масштабирование	Вертикальное	Горизонтальное
Отказоустойчивость	Единая точка отказа	Изолированные отказы
Сложность	Ниже	Выше

#Разделение на сервисы

#По доменам

┌─────────────────────────────────────────┐
│             API Gateway                 │
└─────────────────────────────────────────┘
         │         │         │
         ▼         ▼         ▼
   ┌──────────┐ ┌──────────┐ ┌──────────┐
   │  Users   │ │  Orders  │ │Payments  │
   │ Service  │ │ Service  │ │ Service  │
   └──────────┘ └──────────┘ └──────────┘

#Пример структуры

project/
├── api-gateway/
├── users-service/
├── orders-service/
└── payments-service/

#Коммуникация между сервисами

#Синхронная (HTTP/REST)

import httpx

class OrdersService:
    def __init__(self):
        self.client = httpx.AsyncClient(base_url="http://orders-service:8000")
    
    async def get_order(self, order_id: int):
        response = await self.client.get(f"/orders/{order_id}")
        return response.json()

#Асинхронная (Message Queue)

from aiokafka import AIOKafkaProducer

producer = AIOKafkaProducer(bootstrap_servers='kafka:9092')

await producer.start()
await producer.send('orders', b'{"order_id": 123}')

#API Gateway

from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

@app.get('/api/users/{user_id}')
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://users-service:8000/users/{user_id}")
        if response.status_code == 404:
            raise HTTPException(404, "User not found")
        return response.json()

@app.get('/api/orders/{user_id}')
async def get_orders(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://orders-service:8000/orders?user_id={user_id}")
        return response.json()

#Service Discovery

from consul import Consul

consul = Consul(host='consul', port=8500)

def get_service_url(service_name: str):
    _, services = consul.health.service(service_name, passing_only=True)
    if services:
        service = services[0]['Service']
        return f"http://{service['Address']}:{service['Port']}"
    raise Exception(f"Service {service_name} not found")

#Circuit Breaker

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_external_service():
    async with httpx.AsyncClient() as client:
        response = await client.get("http://external-service/api")
        return response.json()

#Распределённые транзакции

#Saga Pattern

# Order Saga
async def create_order_saga(user_id: int, items: list):
    # Шаг 1: Резервирование товаров
    inventory = await inventory_service.reserve(items)
    
    try:
        # Шаг 2: Создание заказа
        order = await orders_service.create(user_id, items)
        
        # Шаг 3: Оплата
        payment = await payment_service.charge(user_id, order.total)
        
        return order
    except Exception as e:
        # Компенсация: отмена резервирования
        await inventory_service.release(inventory.id)
        raise

#Конфигурация

#Centralized Config

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    users_service_url: str = "http://users-service:8000"
    orders_service_url: str = "http://orders-service:8000"
    kafka_bootstrap_servers: str = "kafka:9092"

#Логирование и трейсинг

from opentelemetry import trace
from opentelemetry.context import attach, set_value

tracer = trace.get_tracer(__name__)

@app.get("/orders/{order_id}")
def get_order(order_id: int):
    with tracer.start_as_current_span("get_order"):
        # Propagate trace ID
        attach(set_value("order_id", order_id))
        
        order = orders_service.get(order_id)
        return order

#Исполняемая лаборатория

Лаборатория «Gateway, timeout, retry, circuit breaker» доступна в публичном GitLab. API вызывает другой сервис через httpx, но без таймаута, без retry, без circuit breaker, а ошибки проксируются с раскрытием внутренних деталей. Нужно реализовать клиент с таймаутом, обработкой 404/5xx, circuit breaker и сокрытием внутренних деталей.

Склонируйте репозиторий и запустите тесты из его корня:

git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest microservices/tests

#Чеклист микросервисов

Критерий

Монолит

Микросервисы

Размер

Одно приложение

Много небольших сервисов

Развёртывание

Одно целое

Независимое

Масштабирование

Вертикальное

Горизонтальное

Отказоустойчивость

Единая точка отказа

Изолированные отказы

Сложность

Ниже

Выше

┌─────────────────────────────────────────┐ │ API Gateway │ └─────────────────────────────────────────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Users │ │ Orders │ │Payments │ │ Service │ │ Service │ │ Service │ └──────────┘ └──────────┘ └──────────┘