Case Study: проектируем SaaS-платформу на Python

# Структура модульного монолита
project/
├── app/
│   ├── tenants/          # Мульти-тенантность
│   ├── imports/          # Импорт данных
│   ├── processing/       # Обработка данных
│   ├── reports/          # Генерация отчётов
│   ├── billing/          # Платежи и подписки
│   ├── notifications/    # Уведомления
│   └── api/              # REST API
├── celery_app.py         # Celery
├── config.py             # Настройки
└── main.py               # FastAPI

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class Tenant(Base):
    __tablename__ = "tenants"

    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    slug = Column(String(100), unique=True, nullable=False)
    plan = Column(String(50), default="free")  # free, pro, enterprise
    is_active = Column(Integer, default=1)

    reports = relationship("Report", back_populates="tenant")

class Report(Base):
    __tablename__ = "reports"

    id = Column(Integer, primary_key=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False)
    title = Column(String(500), nullable=False)
    status = Column(String(30), default="pending")  # pending, processing, done, failed
    file_url = Column(String(1000))
    created_at = Column(String(30), default="2026-01-01")

    tenant = relationship("Tenant", back_populates="reports")

-- Partitioning по tenant_id для больших таблиц
CREATE TABLE report_events (
    id BIGSERIAL,
    tenant_id INTEGER NOT NULL,
    event_type VARCHAR(50),
    payload JSONB,
    created_at TIMESTAMP DEFAULT NOW()
) PARTITION BY LIST (tenant_id);

-- Отдельная партиция для каждого tenant-а (при большом количестве)
CREATE TABLE report_events_tenant_1 PARTITION OF report_events
    FOR VALUES IN (1);

import redis
from fastapi import Request

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# Cache-aside паттерн
async def get_tenant_stats(request: Request, tenant_id: int):
    cache_key = f"tenant_stats:{tenant_id}"

    # Попытка получить из кэша
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Запрос к базе
    stats = await db.get_tenant_stats(tenant_id)

    # Кэширование на 5 минут
    redis_client.setex(cache_key, 300, json.dumps(stats))

    return stats

from celery import Celery

celery_app = Celery(
    "saas_platform",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,  # Подтверждение после выполнения
    worker_prefetch_multiplier=1,
)

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def import_data_task(self, tenant_id: int, file_url: str):
    """Импорт данных из файла"""
    try:
        # Загрузка файла
        data = download_file(file_url)

        # Парсинг и валидация
        parsed = parse_data(data)

        # Сохранение в БД
        save_to_db(tenant_id, parsed)

        # Уведомление
        notify_tenant(tenant_id, "Данные успешно импортированы")

        return {"status": "completed", "rows": len(parsed)}
    except Exception as exc:
        raise self.retry(exc=exc)

@celery_app.task
def generate_report_task(tenant_id: int, report_id: int):
    """Генерация отчёта"""
    report = get_report(report_id)

    # Агрегация данных
    data = aggregate_data(tenant_id, report.filters)

    # Генерация PDF
    pdf_url = generate_pdf(data, report.template)

    # Сохранение результата
    update_report(report_id, status="done", file_url=pdf_url)

    # Уведомление
    notify_tenant(tenant_id, f"Отчёт '{report.title}' готов")

    return {"status": "completed", "url": pdf_url}

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI(title="SaaS Reporting Platform")

# --- Сchemas ---
class ReportCreate(BaseModel):
    title: str
    filters: dict
    template: str

class ReportResponse(BaseModel):
    id: int
    tenant_id: int
    title: str
    status: str
    file_url: str | None = None

# --- Middleware: tenant isolation ---
@app.middleware("http")
async def tenant_middleware(request, call_next):
    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(400, "Missing X-Tenant-ID header")
    request.state.tenant_id = int(tenant_id)
    response = await call_next(request)
    return response

# --- Endpoints ---
@app.post("/reports", response_model=ReportResponse)
async def create_report(
    data: ReportCreate,
    tenant_id: int = Depends(get_tenant_id),
):
    """Создание отчёта (синхронно — быстро)"""
    report = save_report(tenant_id, data)

    # Запуск фоновой генерации
    generate_report_task.delay(tenant_id, report.id)

    return report

@app.get("/reports/{report_id}", response_model=ReportResponse)
async def get_report(
    report_id: int,
    tenant_id: int = Depends(get_tenant_id),
):
    """Статус отчёта"""
    report = get_report_by_id(report_id, tenant_id)
    if not report:
        raise HTTPException(404, "Report not found")
    return report

@app.get("/reports/{report_id}/download")
async def download_report(
    report_id: int,
    tenant_id: int = Depends(get_tenant_id),
):
    """Скачивание готового отчёта"""
    report = get_report_by_id(report_id, tenant_id)
    if report.status != "done":
        raise HTTPException(400, "Report not ready")
    return FileResponse(report.file_url)

@app.post("/imports")
async def import_data(
    file_url: str,
    tenant_id: int = Depends(get_tenant_id),
):
    """Запуск импорта данных"""
    import_data_task.delay(tenant_id, file_url)
    return {"status": "queued"}

from fastapi import Request, HTTPException
import time

def rate_limiter(max_requests: int = 100, window: int = 60):
    async def limiter(request: Request):
        tenant_id = request.state.tenant_id
        key = f"rate:{tenant_id}"

        current = redis_client.get(key)
        if current and int(current) > max_requests:
            raise HTTPException(429, "Rate limit exceeded")

        redis_client.incr(key)
        if not current:
            redis_client.expire(key, window)

    return limiter

# Применение к роутам
app.add_api_route("/reports", create_report, methods=["POST"], dependencies=[Depends(rate_limiter(100, 60))])

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout

    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit is OPEN")

        try:
            result = func(*args, **kwargs)
            self.failure_count = 0
            self.state = CircuitState.CLOSED
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise e

# Использование
cb = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

def send_email(to: str, subject: str, body: str):
    return cb.call(email_service.send, to, subject, body)

import time
import random

def retry_with_backoff(func, max_retries: int = 3, base_delay: float = 1.0):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise e
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)

import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if hasattr(record, "tenant_id"):
            log_data["tenant_id"] = record.tenant_id
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)

# Настройка
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.getLogger("saas_platform").addHandler(handler)

version: "3.8"

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/saas
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

  db:
    image: postgres:16
    environment:
      - POSTGRES_DB=saas
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

  celery-worker:
    build: .
    command: celery -A celery_app worker -l info
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/saas
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

  celery-beat:
    build: .
    command: celery -A celery_app beat -l info
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

volumes:
  pgdata:

┌─────────────────────────────────────────────────────┐
│                    Cloudflare / Nginx               │
│              (SSL, CDN, Rate Limiting)              │
└──────────────────────┬──────────────────────────────┘
                       │
              ┌────────▼────────┐
              │    Load Balancer│
              └────────┬────────┘
                       │
          ┌────────────┴────────────┐
          ▼                          ▼
   ┌─────────────┐          ┌─────────────┐
   │   App 1     │          │   App 2     │
   │  (FastAPI)  │          │  (FastAPI)  │
   └──────┬──────┘          └──────┬──────┘
          │                        │
   ┌──────▼──────┐          ┌──────▼──────┐
   │  Celery W1  │          │  Celery W2  │
   └──────┬──────┘          └──────┬──────┘
          │                        │
   ┌──────▼────────────────────────▼──────┐
   │           Redis (broker + cache)     │
   └──────────────────────────────────────┘
          │
   ┌──────▼──────┐
   │  PostgreSQL │
   │  (Primary)  │
   └─────────────┘
          │
   ┌──────▼──────┐
   │  PostgreSQL │
   │  (Replica)  │
   └─────────────┘

name: Deploy

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: pip install -r requirements.txt -r requirements-dev.txt
      - name: Run tests
        run: pytest -v --cov=app

  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to server
        run: |
          ssh deploy@server "cd /opt/saas && git pull && docker-compose up -d --build"