Микросервис на Python с FastAPI и Docker

from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
class User(BaseModel):
    id: int
    name: str
    email: str
    active: bool = True
 
@app.post("/users/")
async def create_user(user: User):
    # Данные уже проверены и соответствуют модели
    return {"user_id": user.id}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    # Предположим, что db_client — это асинхронный клиент БД
    item = await db_client.get_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    # Параллельный запрос к другому сервису
    related_data = await external_service.get_data(item.external_id)
    
    return {"item": item, "related_data": related_data}

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
@app.get("/users/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    user = authenticate_user(db, token)
    if not user:
        raise HTTPException(status_code=401)
    return user
 
@app.get("/items/")
async def read_items(current_user: User = Depends(get_current_user)):
    return current_user.items

from fastapi import FastAPI
 
app = FastAPI(
    title="Product API",
    description="Микросервис для управления продуктами",
    version="0.1.0"
)
 
@app.get("/")
async def root():
    return {"message": "Добро пожаловать в API продуктов!"}
 
@app.get("/health")
async def health_check():
    return {"status": "healthy"}
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

my_microservice/
├── app/
│   ├── __init__.py
│   ├── main.py              # Точка входа приложения
│   ├── api/                 # Маршруты API
│   │   ├── __init__.py
│   │   ├── routes/          # Маршруты, сгруппированные по функциональности
│   │   │   ├── __init__.py
│   │   │   ├── products.py
│   │   │   └── users.py
│   ├── core/                # Основные компоненты
│   │   ├── __init__.py
│   │   ├── config.py        # Конфигурация приложения
│   │   └── security.py      # Механизмы безопасности
│   ├── db/                  # Код для работы с базой данных
│   │   ├── __init__.py
│   │   ├── models.py        # Модели данных ORM
│   │   └── repository.py    # Абстракции доступа к данным
│   ├── schemas/             # Pydantic-модели для валидации
│   │   ├── __init__.py
│   │   ├── product.py
│   │   └── user.py
│   └── services/            # Бизнес-логика
│       ├── __init__.py
│       └── product_service.py
├── tests/                   # Тесты
│   ├── __init__.py
│   ├── conftest.py
│   └── test_api/
│       ├── __init__.py
│       └── test_products.py
├── .env                     # Переменные окружения
├── .gitignore
├── Dockerfile               # Конфигурация для Docker
├── docker-compose.yml
├── pyproject.toml          # Метаданные проекта и зависимости
└── README.md

from typing import List, Optional
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
 
class Product(BaseModel):
    id: Optional[int] = None
    name: str
    description: Optional[str] = None
    price: float
    available: bool = True
 
router = APIRouter()
 
@router.get("/products/", response_model=List[Product])
async def get_products(
    skip: int = Query(0, ge=0, description="Сколько записей пропустить"),
    limit: int = Query(10, ge=1, le=100, description="Максимальное количество записей")
):
    """
    Получение списка продуктов с пагинацией.
    
    - [B]skip[/B]: Количество записей, которые нужно пропустить
    - [B]limit[/B]: Максимальное количество записей для возврата
    """
    return [
        {"id": 1, "name": "Смартфон", "price": 699.99, "available": True},
        {"id": 2, "name": "Ноутбук", "price": 1299.99, "available": False}
    ]

from fastapi import FastAPI, HTTPException, Path
from pydantic import BaseModel, Field, validator
 
app = FastAPI()
 
class ProductCreate(BaseModel):
    name: str = Field(..., min_length=3, max_length=50)
    description: str = Field(None, max_length=1000)
    price: float = Field(..., gt=0)
    category_id: int
    
    @validator('name')
    def name_must_be_valid(cls, v):
        if v.strip() == "":
            raise ValueError('Название не может быть пустым')
        return v.strip()
    
    class Config:
        schema_extra = {
            "example": {
                "name": "Игровая мышь",
                "description": "Высокоточная игровая мышь с RGB подсветкой",
                "price": 59.99,
                "category_id": 5
            }
        }
 
@app.post("/products/", response_model=ProductCreate)
async def create_product(product: ProductCreate):
    # В реальном приложении здесь был бы код для сохранения в БД
    return product

@app.get("/products/{product_id}")
async def get_product(product_id: int = Path(..., gt=0)):
    # Имитация поиска в базе данных
    product = None  # предположим, что продукт не найден
    
    if not product:
        raise HTTPException(
            status_code=404,
            detail=f"Продукт с ID {product_id} не найден",
            headers={"X-Error": "PRODUCT_NOT_FOUND"},
        )
    
    return {"id": product_id, "name": "Тестовый продукт"}

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
app = FastAPI()
 
class CustomException(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
 
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message}
    )
 
@app.get("/error-test")
async def error_test():
    raise CustomException(code="BUSINESS_RULE_VIOLATED", message="Невозможно выполнить операцию")

from fastapi import APIRouter, Depends
from app.dependencies import get_token_header
 
router = APIRouter(
    prefix="/products",
    tags=["products"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Не найдено"}},
)
 
@router.get("/")
async def read_products():
    return [{"name": "Часы", "price": 49.99}, {"name": "Наушники", "price": 99.99}]
 
@router.get("/{product_id}")
async def read_product(product_id: int):
    return {"name": "Часы", "product_id": product_id}

from fastapi import FastAPI
from app.api.routes import products, users
 
app = FastAPI()
app.include_router(products.router)
app.include_router(users.router)

from typing import List
from fastapi import APIRouter, Query
from pydantic import BaseModel
 
router = APIRouter()
 
class Product(BaseModel):
    id: int
    name: str
    price: float
 
@router.get("/products/", response_model=List[Product])
async def get_products(
    offset: int = Query(0, ge=0, description="Смещение от начала списка"),
    limit: int = Query(10, ge=1, le=100, description="Максимальное число элементов")
):
    # Здесь был бы код получения данных из БД с указанными параметрами
    products = [{"id": i, "name": f"Продукт {i}", "price": float(i * 10)} for i in range(offset, offset + limit)]
    return products

class PaginatedResponse(BaseModel):
    items: List[Product]
    total: int
    next_cursor: str = None
 
@router.get("/products/cursor", response_model=PaginatedResponse)
async def get_products_cursor(
    cursor: str = None,
    limit: int = Query(10, ge=1, le=100)
):
    # Декодирование курсора для получения последнего ID
    last_id = 0 if not cursor else int(cursor)
    
    # Имитация получения данных из БД
    items = [{"id": i, "name": f"Продукт {i}", "price": float(i * 10)} 
             for i in range(last_id + 1, last_id + limit + 1)]
    
    next_cursor = str(last_id + limit) if items else None
    
    return {
        "items": items, 
        "total": 100,  # общее число элементов
        "next_cursor": next_cursor
    }

from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional
 
class Address(BaseModel):
    street: str
    city: str
    zip_code: str = Field(..., regex=r"^\d{5}(-\d{4})?$")
    
class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None
    addresses: List[Address] = []
    
@app.post("/users/")
async def create_user(user: User):
    return {"message": "Пользователь создан", "user": user}

.git
.github
venv
__pycache__
*.pyc
.pytest_cache
.coverage
htmlcov
tests
docs

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy import Column, Integer, String, Float, Boolean
 
# Настройка подключения
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
# Определение модели данных
Base = declarative_base()
 
class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    price = Column(Float)
    available = Column(Boolean, default=True)
 
# Использование в FastAPI
@app.get("/products/{product_id}")
async def get_product(product_id: int):
    async with async_session() as session:
        query = select(Product).where(Product.id == product_id)
        result = await session.execute(query)
        product = result.scalars().first()
        
        if not product:
            raise HTTPException(status_code=404, detail="Продукт не найден")
        
        return product

from motor.motor_asyncio import AsyncIOMotorClient
 
# Подключение
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.product_db
collection = db.products
 
# Использование в FastAPI
@app.post("/products/")
async def create_product(product: ProductCreate):
    product_dict = product.dict()
    result = await collection.insert_one(product_dict)
    
    created_product = await collection.find_one({"_id": result.inserted_id})
    return created_product

class ProductRepository:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    async def get_by_id(self, product_id: int):
        async with self.session_factory() as session:
            query = select(Product).where(Product.id == product_id)
            result = await session.execute(query)
            return result.scalars().first()
    
    async def create(self, product_data: dict):
        product = Product(**product_data)
        async with self.session_factory() as session:
            session.add(product)
            await session.commit()
            await session.refresh(product)
            return product

import asyncio
from fastapi import APIRouter
 
router = APIRouter()
 
async def fetch_product_details(product_id: int):
    # Имитация задержки при запросе данных
    await asyncio.sleep(1)
    return {"id": product_id, "name": f"Продукт {product_id}"}
 
async def fetch_product_reviews(product_id: int):
    # Имитация задержки при запросе отзывов
    await asyncio.sleep(0.5)
    return [
        {"user": "user1", "rating": 5, "comment": "Отличный продукт!"},
        {"user": "user2", "rating": 4, "comment": "Хороший выбор"}
    ]
 
@router.get("/products/{product_id}/full")
async def get_product_with_reviews(product_id: int):
    # Параллельное выполнение двух асинхронных операций
    product_details, reviews = await asyncio.gather(
        fetch_product_details(product_id),
        fetch_product_reviews(product_id)
    )
    
    return {
        "product": product_details,
        "reviews": reviews
    }

from fastapi import BackgroundTasks
 
@app.post("/orders/")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
    # Быстрая операция - сохранение заказа
    order_id = await save_order(order)
    
    # Медленные операции выполняются в фоне
    background_tasks.add_task(process_payment, order_id, order.payment_details)
    background_tasks.add_task(send_order_confirmation, order.user_email, order_id)
    
    return {"order_id": order_id, "status": "processing"}

# test_product_service.py
import pytest
from app.services.product_service import ProductService
from app.repositories.product_repository import ProductRepository
 
class MockProductRepository:
    async def get_by_id(self, product_id: int):
        if product_id == 1:
            return {"id": 1, "name": "Test Product", "price": 9.99}
        return None
 
@pytest.fixture
def product_service():
    repository = MockProductRepository()
    return ProductService(repository)
 
@pytest.mark.asyncio
async def test_get_product_returns_product_when_exists(product_service):
    # Действие
    product = await product_service.get_product(1)
    
    # Проверка
    assert product["id"] == 1
    assert product["name"] == "Test Product"
 
@pytest.mark.asyncio
async def test_get_product_raises_exception_when_not_exists(product_service):
    # Проверка исключения
    with pytest.raises(ValueError, match="Продукт не найден"):
        await product_service.get_product(999)

# test_api.py
from fastapi.testclient import TestClient
from app.main import app
 
client = TestClient(app)
 
def test_read_product():
    response = client.get("/products/1")
    assert response.status_code == 200
    assert response.json()["name"] == "Test Product"
 
def test_create_product():
    product_data = {
        "name": "New Product",
        "price": 29.99,
        "description": "Test description"
    }
    response = client.post("/products/", json=product_data)
    assert response.status_code == 201
    assert response.json()["name"] == "New Product"

import pytest
from unittest.mock import AsyncMock
 
@pytest.fixture
async def mock_db_session():
    session = AsyncMock()
    session.__aenter__.return_value = session
    
    # Мок для запроса
    mock_result = AsyncMock()
    mock_result.scalars.return_value.first.return_value = {
        "id": 1,
        "name": "Test Product"
    }
    session.execute.return_value = mock_result
    
    yield session
 
@pytest.mark.asyncio
async def test_get_product_from_db(mock_db_session):
    repository = ProductRepository(lambda: mock_db_session)
    product = await repository.get_by_id(1)
    
    # Проверяем, что вызов execute был с правильными параметрами
    mock_db_session.execute.assert_called_once()
    assert product["id"] == 1

import logging
from fastapi import FastAPI, Request
import time
from typing import Callable
import uuid
 
# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)
 
logger = logging.getLogger(__name__)
 
app = FastAPI()
 
# Middleware для логирования запросов
@app.middleware("http")
async def log_requests(request: Request, call_next: Callable):
    request_id = str(uuid.uuid4())
    logger.info(f"Request {request_id} started: {request.method} {request.url}")
    
    start_time = time.time()
    try:
        response = await call_next(request)
        process_time = time.time() - start_time
        logger.info(f"Request {request_id} completed: {response.status_code} in {process_time:.3f}s")
        
        # Добавляем ID запроса в заголовки ответа для отслеживания
        response.headers["X-Request-ID"] = request_id
        return response
    except Exception as e:
        logger.error(f"Request {request_id} failed: {str(e)}")
        raise

import json
import logging
import datetime
 
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            "timestamp": datetime.datetime.now().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }
        
        if hasattr(record, 'request_id'):
            log_record["request_id"] = record.request_id
        
        if record.exc_info:
            log_record["exception"] = self.formatException(record.exc_info)
            
        return json.dumps(log_record)

from prometheus_client import Counter, Histogram
from prometheus_fastapi_instrumentator import Instrumentator
 
# Инициализация Instrumentator
instrumentator = Instrumentator().instrument(app)
 
# Настройка собственных метрик
http_requests_total = Counter(
    'http_requests_total', 
    'Total HTTP requests', 
    ['method', 'endpoint', 'status']
)
 
http_request_duration = Histogram(
    'http_request_duration_seconds', 
    'HTTP request duration in seconds',
    ['method', 'endpoint']
)
 
@app.middleware("http")
async def monitor_requests(request: Request, call_next: Callable):
    method = request.method
    path = request.url.path
    
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    
    status = response.status_code
    http_requests_total.labels(method=method, endpoint=path, status=status).inc()
    http_request_duration.labels(method=method, endpoint=path).observe(duration)
    
    return response
 
# Активация инструментации
@app.on_event("startup")
async def startup():
    instrumentator.expose(app)

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel
 
# Настройки безопасности
SECRET_KEY = "ваш_секретный_ключ"  # В production берите из переменных окружения!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
# Модели
class Token(BaseModel):
    access_token: str
    token_type: str
 
class TokenData(BaseModel):
    username: str = None
 
class User(BaseModel):
    username: str
    email: str = None
    disabled: bool = False
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
# Функции для работы с JWT
def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt
 
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Неверные учетные данные",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    
    # Здесь должна быть проверка пользователя в БД
    user = {"username": token_data.username, "disabled": False}
    return user
 
# Эндпоинт для аутентификации
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # Проверка пользователя (заглушка)
    if form_data.username != "test" or form_data.password != "test":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверное имя пользователя или пароль",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}
 
# Защищенный эндпоинт
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

import redis
from fastapi import FastAPI, Depends
import json
import pickle
from functools import wraps
 
# Инициализация Redis-клиента
redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
app = FastAPI()
 
# Декоратор для кэширования результатов функций
def cache(expire_seconds: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Формирование ключа кэша из параметров функции
            key = f"cache:{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # Проверка наличия данных в кэше
            cached_data = redis_client.get(key)
            if cached_data:
                return pickle.loads(cached_data)
            
            # Выполнение функции и сохранение результата в кэш
            result = await func(*args, **kwargs)
            redis_client.setex(key, expire_seconds, pickle.dumps(result))
            return result
        return wrapper
    return decorator
 
# Применение кэширования к эндпоинту
@app.get("/products/{product_id}")
@cache(expire_seconds=300)  # Кэш на 5 минут
async def get_product(product_id: int):
    # Имитация медленной операции получения данных
    # В реальном приложении здесь был бы запрос к БД
    await asyncio.sleep(1)
    return {"id": product_id, "name": f"Продукт {product_id}", "price": 100.0}

from cachetools import TTLCache
from datetime import timedelta
 
# Создание локального кэша с временем жизни элементов 5 минут
local_cache = TTLCache(maxsize=100, ttl=timedelta(minutes=5).total_seconds())
 
def get_cached_or_fetch(key, fetch_func):
    # Проверка наличия в кэше
    if key in local_cache:
        return local_cache[key]
    
    # Получение и сохранение в кэш
    result = fetch_func()
    local_cache[key] = result
    return result

import httpx
 
async def get_user_orders(user_id: int):
  async with httpx.AsyncClient() as client:
      # Запрос к сервису заказов
      response = await client.get(f"http://order-service/api/orders?user_id={user_id}")
      response.raise_for_status()
      return response.json()
 
@app.get("/users/{user_id}/dashboard")
async def get_user_dashboard(user_id: int):
  # Получение данных пользователя из локальной БД
  user = await user_repository.get_by_id(user_id)
  if not user:
      raise HTTPException(status_code=404, detail="Пользователь не найден")
  
  # Получение заказов пользователя из другого сервиса
  orders = await get_user_orders(user_id)
  
  return {
      "user": user,
      "orders": orders,
      "orders_count": len(orders)
  }

class ServiceClient:
  def __init__(self, base_url: str, timeout: float = 5.0, retries: int = 3):
      self.base_url = base_url
      self.timeout = timeout
      self.retries = retries
  
  async def request(self, method: str, path: str, **kwargs):
      url = f"{self.base_url}{path}"
      
      # Настройка таймаута и повторных попыток
      kwargs.setdefault("timeout", self.timeout)
      
      # Применение паттерна Circuit Breaker для защиты от каскадных сбоев
      for attempt in range(self.retries):
          try:
              async with httpx.AsyncClient() as client:
                  response = await client.request(method, url, **kwargs)
                  response.raise_for_status()
                  return response.json()
          except (httpx.RequestError, httpx.HTTPStatusError) as e:
              if attempt == self.retries - 1:
                  # Последняя попытка, пробрасываем исключение выше
                  raise ServiceCommunicationError(f"Ошибка при обращении к сервису: {str(e)}")
              # Экспоненциальная задержка перед повторной попыткой
              await asyncio.sleep(0.1 * (2 ** attempt))

import aio_pika
import json
 
# Настройка подключения к RabbitMQ
async def get_rabbitmq_connection():
  return await aio_pika.connect_robust("amqp://guest:guest@rabbitmq:5672/")
 
# Отправка сообщения
async def send_event(event_type: str, payload: dict):
  connection = await get_rabbitmq_connection()
  async with connection:
      channel = await connection.channel()
      
      # Объявление обменника для событий
      exchange = await channel.declare_exchange(
          "events", aio_pika.ExchangeType.TOPIC
      )
      
      # Создание и отправка сообщения
      message = aio_pika.Message(
          body=json.dumps(payload).encode(),
          content_type="application/json",
          headers={"event_type": event_type}
      )
      
      await exchange.publish(message, routing_key=event_type)
 
# Обработчик эндпоинта, отправляющий событие
@app.post("/products/")
async def create_product(product: ProductCreate):
  # Создание продукта в БД
  db_product = await product_repository.create(product.dict())
  
  # Отправка события о создании продукта
  await send_event(
      "product.created",
      {"product_id": db_product.id, "name": db_product.name}
  )
  
  return db_product

async def process_messages():
  connection = await get_rabbitmq_connection()
  async with connection:
      channel = await connection.channel()
      
      # Объявление обменника
      exchange = await channel.declare_exchange(
          "events", aio_pika.ExchangeType.TOPIC
      )
      
      # Создание очереди для данного сервиса
      queue = await channel.declare_queue("inventory-service-queue", durable=True)
      
      # Подписка на события определенного типа
      await queue.bind(exchange, routing_key="product.created")
      await queue.bind(exchange, routing_key="product.updated")
      
      # Начать получение сообщений
      async with queue.iterator() as queue_iter:
          async for message in queue_iter:
              async with message.process():
                  try:
                      body = json.loads(message.body.decode())
                      event_type = message.headers["event_type"]
                      
                      if event_type == "product.created":
                          await handle_product_created(body)
                      elif event_type == "product.updated":
                          await handle_product_updated(body)
                  except Exception as e:
                      # Логирование ошибки и возможное перенаправление в очередь ошибок
                      logging.error(f"Ошибка обработки сообщения: {str(e)}")
 
# Запуск обработчика сообщений при старте приложения
@app.on_event("startup")
async def startup_event_handlers():
  # Запуск фоновой задачи для обработки сообщений
  asyncio.create_task(process_messages())

# Модель события
class Event(BaseModel):
  id: UUID = Field(default_factory=uuid4)
  timestamp: datetime = Field(default_factory=datetime.utcnow)
  type: str
  data: dict
  metadata: dict = {}
 
# Сервис для работы с событиями
class EventStore:
  def __init__(self, mongo_client):
      self.db = mongo_client.get_database("event_store")
      self.events = self.db.get_collection("events")
  
  async def save_event(self, event: Event):
      await self.events.insert_one(event.dict())
      # Публикация события в шину сообщений
      await publish_event(event)
  
  async def get_events(self, event_type: str = None, start_date: datetime = None):
      query = {}
      if event_type:
          query["type"] = event_type
      if start_date:
          query["timestamp"] = {"$gte": start_date}
      
      cursor = self.events.find(query).sort("timestamp", 1)
      return [Event(**event) async for event in cursor]
 
# Использование в обработчике API
@app.post("/orders/")
async def create_order(order: OrderCreate):
  # Создание события
  event = Event(
      type="OrderCreated",
      data=order.dict()
  )
  
  # Сохранение события
  await event_store.save_event(event)
  
  return {"order_id": event.data["id"], "status": "processing"}

from databases import Database
from sqlalchemy.ext.declarative import declarative_base
 
DATABASE_URL = "postgresql://user:password@localhost/dbname"
database = Database(DATABASE_URL)
Base = declarative_base()
 
# В FastAPI приложении
app = FastAPI()
 
@app.on_event("startup")
async def startup():
  await database.connect()
 
@app.on_event("shutdown")
async def shutdown():
  await database.disconnect()

from fastapi import BackgroundTasks
 
@app.post("/reports/generate")
async def generate_report(report_params: ReportParams, background_tasks: BackgroundTasks):
  # Быстрый ответ клиенту
  report_id = str(uuid4())
  
  # Запуск генерации отчета в фоне
  background_tasks.add_task(generate_report_task, report_id, report_params)
  
  return {"report_id": report_id, "status": "processing"}

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi_sessions.backends.redis import RedisBackend
from fastapi_sessions.session_verifier import SessionVerifier
from fastapi_sessions.frontends.cookie import CookieParameters, SessionCookie
from redis.asyncio import Redis
 
redis = Redis(host="redis", port=6379)
backend = RedisBackend(redis, prefix="sessions:")
 
cookie_params = CookieParameters()
cookie = SessionCookie(
  cookie_name="session",
  identifier="general_verifier",
  auto_error=True,
  secret_key="SECURE_KEY_HERE",
  cookie_params=cookie_params
)
 
# Определение данных сессии
class SessionData(BaseModel):
  user_id: int
 
# Верификатор сессии
class BasicVerifier(SessionVerifier[SessionData]):
  def __init__(self, backend: RedisBackend, name: str):
      self.backend = backend
      self.name = name
      
  async def read_data(self, session_id: str) -> SessionData:
      data = await self.backend.read(session_id)
      if not data:
          raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
      return SessionData(**data)
  
  async def verify_session(self, session_id: str) -> bool:
      exists = await self.backend.exists(session_id)
      return exists
 
verifier = BasicVerifier(backend=backend, name="general_verifier")
 
# Использование в API
@app.get("/users/me")
async def get_current_user(session_data: SessionData = Depends(cookie)):
  return {"user_id": session_data.user_id}

import redis.asyncio as redis
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
 
@app.on_event("startup")
async def startup():
  redis_client = redis.Redis(host="redis", port=6379)
  FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache:")
 
@app.get("/products/{product_id}")
@cache(expire=60)  # Кэширование на 60 секунд
async def get_product(product_id: int):
  # Медленная операция получения данных
  return await product_repository.get_by_id(product_id)

from celery import Celery
from fastapi import FastAPI
 
# Инициализация Celery
celery_app = Celery(
  "tasks",
  broker="amqp://guest:guest@rabbitmq:5672//",
  backend="redis://redis:6379/0"
)
 
# Определение задачи
@celery_app.task
def process_order(order_id: str):
  # Длительная обработка заказа
  pass
 
app = FastAPI()
 
@app.post("/orders/")
async def create_order(order: OrderCreate):
  # Сохранение заказа в БД
  order_id = await save_order(order)
  
  # Запуск асинхронной обработки
  task = process_order.delay(order_id)
  
  return {"order_id": order_id, "task_id": task.id}
 
@app.get("/orders/{order_id}/status")
async def get_order_status(order_id: str, task_id: str):
  # Проверка статуса задачи
  task = process_order.AsyncResult(task_id)
  if task.state == 'PENDING':
      return {"status": "processing"}
  elif task.state == 'SUCCESS':
      return {"status": "completed"}
  else:
      return {"status": "failed", "error": str(task.result)}