УРОВЕНЬ 0: ЯДРО
├── 0.1 Фреймворк
├── 0.2 Сервер
└── 0.3 Конфигурация

УРОВЕНЬ 1: ДАННЫЕ
├── 1.1 ORM / База данных
├── 1.2 Миграции
├── 1.3 Валидация
├── 1.4 Сериализация
└── 1.5 Кэширование

УРОВЕНЬ 2: БЕЗОПАСНОСТЬ
├── 2.1 Хеширование паролей
├── 2.2 JWT токены
├── 2.3 OAuth2
└── 2.4 CORS / Security headers

УРОВЕНЬ 3: БИЗНЕС-ЛОГИКА
├── 3.1 Фоновые задачи
├── 3.2 Планировщик
├── 3.3 Очереди сообщений
└── 3.4 State machines

УРОВЕНЬ 4: ИНТЕГРАЦИИ
├── 4.1 HTTP клиент
├── 4.2 Email
├── 4.3 SMS
├── 4.4 Платежи
└── 4.5 Внешние API

УРОВЕНЬ 5: ФАЙЛЫ
├── 5.1 Файловая система
├── 5.2 S3 / Облако
├── 5.3 Изображения
├── 5.4 Excel / CSV
└── 5.5 PDF

УРОВЕНЬ 6: REAL-TIME
├── 6.1 WebSocket
├── 6.2 SSE
└── 6.3 Pub/Sub

УРОВЕНЬ 7: UI / FRONTEND
├── 7.1 Шаблоны
├── 7.2 Статика
└── 7.3 Админка

УРОВЕНЬ 8: ТЕСТИРОВАНИЕ
├── 8.1 Фреймворк
├── 8.2 Моки
├── 8.3 Фикстуры
└── 8.4 Покрытие

УРОВЕНЬ 9: OBSERVABILITY
├── 9.1 Логирование
├── 9.2 Метрики
├── 9.3 Трейсинг
└── 9.4 Профилирование

УРОВЕНЬ 10: ИНФРАСТРУКТУРА
├── 10.1 Контейнеризация
├── 10.2 ASGI сервер
├── 10.3 Reverse proxy
└── 10.4 CI/CD

УРОВЕНЬ 11: DEV TOOLS
├── 11.1 Линтер
├── 11.2 Форматтер
├── 11.3 Type checker
└── 11.4 Package manager

# Почему FastAPI:
# 1. В 10-20x быстрее Django/Flask
# 2. Автоматическая OpenAPI документация
# 3. Валидация через Pydantic
# 4. Async из коробки
# 5. Type hints = меньше багов

from fastapi import FastAPI
app = FastAPI()

# Development
uvicorn app.main:app --reload

# Production
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker

# Почему pydantic-settings:
# 1. Валидация типов (int, bool, URL)
# 2. Значения по умолчанию
# 3. Документация настроек
# 4. Интеграция с FastAPI

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    debug: bool = False
    api_key: str

    class Config:
        env_file = ".env"

settings = Settings()  # Автоматически загрузит из .env

# Почему SQLAlchemy 2.0:
# 1. Async native (не через greenlet)
# 2. Гибкий query builder
# 3. Поддержка всех БД
# 4. Type hints
# 5. Огромное сообщество

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)

# Почему Beanie для MongoDB:
# 1. Pydantic модели = валидация
# 2. Async native
# 3. Миграции из коробки

from beanie import Document

class User(Document):
    email: str
    name: str

    class Settings:
        collection = "users"

# Почему Alembic:
# 1. Автогенерация миграций из моделей
# 2. Откат (downgrade)
# 3. Версионирование
# 4. Работает с SQLAlchemy

# Команды:
# alembic init alembic
# alembic revision --autogenerate -m "add users"
# alembic upgrade head
# alembic downgrade -1

# Почему Pydantic v2:
# 1. В 5-50x быстрее v1
# 2. Type hints = документация
# 3. Автоматическая валидация в FastAPI
# 4. JSON Schema генерация

from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8)
    age: int = Field(ge=18, le=120)

# Почему orjson:
# 1. В 10x быстрее stdlib
# 2. Поддержка datetime, UUID, numpy
# 3. Строгий (не сериализует что попало)

import orjson

data = {"name": "test", "created": datetime.now()}
json_bytes = orjson.dumps(data)  # bytes
json_str = json_bytes.decode()   # str

# Почему Redis:
# 1. Персистентность (данные не теряются)
# 2. Pub/Sub для real-time
# 3. Структуры данных (lists, sets, hashes)
# 4. Кластеризация

import redis.asyncio as redis

r = redis.from_url("redis://localhost")
await r.set("key", "value", ex=3600)  # TTL 1 час
value = await r.get("key")

# Почему passlib:
# 1. Абстракция над алгоритмами
# 2. Автоматический upgrade хешей
# 3. Проверенный в production

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

hashed = pwd_context.hash("password123")
is_valid = pwd_context.verify("password123", hashed)

# Почему python-jose:
# 1. Поддержка RSA, EC ключей
# 2. JWS, JWE, JWK
# 3. Рекомендован FastAPI

from jose import jwt

token = jwt.encode({"sub": user_id}, SECRET_KEY, algorithm="HS256")
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

# FastAPI встроенный OAuth2:
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    # Проверка, выдача токена
    pass

@app.get("/protected")
async def protected(token: str = Depends(oauth2_scheme)):
    # Защищённый endpoint
    pass

ДЕРЕВО ВЫБОРА:

Задача < 5 сек, не критична?
├── ДА → BackgroundTasks
│
└── НЕТ → Нужен retry/статус?
          ├── НЕТ → BackgroundTasks
          │
          └── ДА → I/O или CPU?
                   ├── I/O → ARQ
                   └── CPU → Celery

# BackgroundTasks (простое)
from fastapi import BackgroundTasks

@app.post("/send-email")
async def send(bg: BackgroundTasks):
    bg.add_task(send_email, "user@example.com")
    return {"status": "queued"}

# ARQ (async задачи)
async def send_email(ctx, email: str):
    await smtp.send(email)

class WorkerSettings:
    functions = [send_email]
    redis_settings = RedisSettings()

# Почему APScheduler:
# 1. Работает внутри приложения
# 2. Cron синтаксис
# 3. Персистентность (БД)

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

@scheduler.scheduled_job('cron', hour=3)
async def nightly_task():
    await cleanup_old_data()

scheduler.start()

# Почему transitions:
# 1. Явные состояния и переходы
# 2. Callbacks на переходы
# 3. Валидация переходов

from transitions import Machine

class Order:
    states = ['new', 'paid', 'shipped', 'delivered', 'cancelled']

    def __init__(self):
        self.machine = Machine(model=self, states=Order.states, initial='new')
        self.machine.add_transition('pay', 'new', 'paid')
        self.machine.add_transition('ship', 'paid', 'shipped')
        self.machine.add_transition('deliver', 'shipped', 'delivered')
        self.machine.add_transition('cancel', ['new', 'paid'], 'cancelled')

order = Order()
order.pay()      # new → paid
order.ship()     # paid → shipped

# Почему httpx:
# 1. Async native
# 2. HTTP/2 поддержка
# 3. API как у requests
# 4. Type hints

import httpx

# Async
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com")

# Sync (для скриптов)
response = httpx.get("https://api.example.com")

# Почему tenacity:
# 1. Декоратор = минимум кода
# 2. Exponential backoff
# 3. Условия retry

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def call_api():
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

# Почему fastapi-mail:
# 1. Jinja2 шаблоны для писем
# 2. Вложения
# 3. Async

from fastapi_mail import FastMail, MessageSchema, ConnectionConfig

conf = ConnectionConfig(
    MAIL_USERNAME="user",
    MAIL_PASSWORD="pass",
    MAIL_FROM="noreply@example.com",
    MAIL_SERVER="smtp.example.com"
)

message = MessageSchema(
    subject="Welcome",
    recipients=["user@example.com"],
    template_body={"name": "User"}
)

fm = FastMail(conf)
await fm.send_message(message, template_name="welcome.html")

# Stripe
import stripe
stripe.api_key = "sk_..."

payment_intent = stripe.PaymentIntent.create(
    amount=1000,  # в копейках
    currency="rub"
)

# YooKassa
from yookassa import Payment

payment = Payment.create({
    "amount": {"value": "100.00", "currency": "RUB"},
    "confirmation": {"type": "redirect", "return_url": "https://..."},
    "description": "Заказ #123"
})

# Почему aiofiles:
# 1. Async чтение/запись
# 2. Не блокирует

import aiofiles
from pathlib import Path

async def read_file(path: Path) -> str:
    async with aiofiles.open(path) as f:
        return await f.read()

# Почему aioboto3:
# 1. Async операции
# 2. Полный API AWS

import aioboto3

session = aioboto3.Session()
async with session.client("s3") as s3:
    await s3.upload_file("local.txt", "bucket", "remote.txt")

# Pillow — всё что нужно для изображений
from PIL import Image

img = Image.open("photo.jpg")
img.thumbnail((800, 600))
img.save("thumbnail.jpg", quality=85)

# openpyxl для Excel
from openpyxl import load_workbook

wb = load_workbook("data.xlsx")
ws = wb.active
for row in ws.iter_rows(values_only=True):
    print(row)

# FastAPI WebSocket
from fastapi import WebSocket

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Received: {data}")

from sse_starlette.sse import EventSourceResponse

@app.get("/stream")
async def stream():
    async def event_generator():
        while True:
            yield {"data": "message"}
            await asyncio.sleep(1)
    return EventSourceResponse(event_generator())

<!-- htmx — AJAX без JavaScript -->
<button hx-post="/api/like" hx-swap="outerHTML">
    Like
</button>

<!-- Alpine.js — реактивность -->
<div x-data="{ count: 0 }">
    <button @click="count++">+</button>
    <span x-text="count"></span>
</div>

# SQLAdmin
from sqladmin import Admin, ModelView

admin = Admin(app, engine)

class UserAdmin(ModelView, model=User):
    column_list = [User.id, User.email, User.created_at]
    column_searchable_list = [User.email]

admin.add_view(UserAdmin)

# Полный пример теста
import pytest
from httpx import AsyncClient, ASGITransport

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/users", json={"email": "test@test.com"})
    assert response.status_code == 200

# loguru — для разработки
from loguru import logger
logger.info("Processing {count} items", count=42)

# structlog — для production
import structlog
log = structlog.get_logger()
log.info("processing", count=42, user_id=123)
# Вывод: {"event": "processing", "count": 42, "user_id": 123, "timestamp": "..."}

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()
Instrumentator().instrument(app).expose(app)

# Метрики доступны на /metrics

# pyproject.toml
[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]

# uv — новый стандарт
uv pip install fastapi
uv pip compile requirements.in -o requirements.txt
uv venv