FastAPI Кодер

# СТАНДАРТ: async SQLAlchemy 2.0
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("postgresql+asyncpg://...")

class Base(DeclarativeBase):
    pass

# СТАНДАРТ: JWT + OAuth2
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401)

# СТАНДАРТ: SQLAdmin
from sqladmin import Admin, ModelView
from fastapi import FastAPI

app = FastAPI()
admin = Admin(app, engine)

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

admin.add_view(UserAdmin)

ВЫБОР:

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

# СТАНДАРТ: ARQ для async задач
from arq import create_pool
from arq.connections import RedisSettings

async def send_email(ctx, email: str, message: str):
    # async отправка
    await smtp.send(email, message)

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

# СТАНДАРТ: Decimal для цен
from decimal import Decimal
from pydantic import BaseModel, condecimal

class Price(BaseModel):
    amount: condecimal(max_digits=10, decimal_places=2)
    currency: str = "RUB"

# СТАНДАРТ: httpx + tenacity
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_external_api(url: str):
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

# СТАНДАРТ: WebSocket
from fastapi import WebSocket

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

# СТАНДАРТ: async upload + S3
from fastapi import UploadFile
import aioboto3

async def upload_to_s3(file: UploadFile, bucket: str):
    session = aioboto3.Session()
    async with session.client("s3") as s3:
        await s3.upload_fileobj(file.file, bucket, file.filename)

# СТАНДАРТ: pytest + httpx
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_read_main(client):
    response = await client.get("/")
    assert response.status_code == 200

# СТАНДАРТ: Prometheus
from prometheus_fastapi_instrumentator import Instrumentator

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

# СТАНДАРТ: Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["gunicorn", "app.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]

app/
├── main.py              ← Точка входа, создание app
├── config.py            ← Settings (pydantic-settings)
├── dependencies.py      ← DI: get_db, get_current_user
│
├── routers/             ← Эндпоинты по доменам
│   ├── __init__.py
│   ├── users.py
│   ├── products.py
│   └── orders.py
│
├── schemas/             ← Pydantic модели (DTO)
│   ├── __init__.py
│   ├── user.py          ← UserCreate, UserRead, UserUpdate
│   └── product.py
│
├── models/              ← SQLAlchemy модели (ORM)
│   ├── __init__.py
│   ├── user.py
│   └── product.py
│
├── crud/                ← Операции с БД
│   ├── __init__.py
│   ├── base.py          ← CRUDBase[Model]
│   └── user.py
│
├── services/            ← Бизнес-логика
│   ├── __init__.py
│   ├── email.py
│   └── payment.py
│
├── core/                ← Ядро
│   ├── security.py      ← JWT, хеширование
│   └── exceptions.py    ← Кастомные исключения
│
├── tasks/               ← Фоновые задачи (ARQ/Celery)
│   ├── __init__.py
│   └── email_tasks.py
│
└── tests/
    ├── conftest.py      ← Фикстуры
    ├── test_users.py
    └── test_products.py

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
YYYY-MM-DD | 🚀 FastAPI | {режим} | {задача}
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

[результат]