API на базе FastAPI с Python за пару минут

fastapi_app/
├── venv/                # Виртуальное окружение
├── main.py              # Главный файл приложения

from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/")
def read_root():
    return {"message": "Привет, FastAPI!"}

from fastapi import FastAPI, HTTPException
 
app = FastAPI()
 
# Имитация базы данных
products = []
 
@app.post("/products/")
def create_product(name: str, price: float):
    product = {"id": len(products), "name": name, "price": price}
    products.append(product)
    return product
 
@app.get("/products/{product_id}")
def read_product(product_id: int):
    if product_id < 0 or product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    return products[product_id]

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
 
class Product(BaseModel):
    name: str
    price: float
    description: str = None  # Опциональное поле
 
app = FastAPI()
 
products = []
 
@app.post("/products/")
def create_product(product: Product):
    new_product = product.dict()
    new_product["id"] = len(products)
    products.append(new_product)
    return new_product
 
@app.get("/products/{product_id}")
def read_product(product_id: int):
    if product_id < 0 or product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    return products[product_id]

@app.get("/products/")
def list_products(skip: int = 0, limit: int = 10, min_price: float = None):
    result = products
    
    # Фильтрация по минимальной цене
    if min_price is not None:
        result = [p for p in result if p["price"] >= min_price]
    
    # Пагинация
    return result[skip : skip + limit]

from fastapi import FastAPI, HTTPException, Query, Path
 
@app.get("/products/")
def list_products(
    skip: int = Query(0, ge=0, description="Сколько продуктов пропустить"),
    limit: int = Query(10, ge=1, le=100, description="Максимальное количество продуктов"),
    min_price: float = Query(None, description="Минимальная цена для фильтрации")
):
    result = products
    
    if min_price is not None:
        result = [p for p in result if p["price"] >= min_price]
    
    return result[skip : skip + limit]
 
@app.get("/products/{product_id}")
def read_product(product_id: int = Path(..., ge=0, description="ID продукта")):
    if product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    return products[product_id]

@app.put("/products/{product_id}")
def update_product(product_id: int, product: Product):
    if product_id < 0 or product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    
    updated_product = product.dict()
    updated_product["id"] = product_id
    products[product_id] = updated_product
    return updated_product
 
@app.delete("/products/{product_id}")
def delete_product(product_id: int):
    if product_id < 0 or product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    
    deleted_product = products.pop(product_id)
    
    # Обновляем ID оставшихся продуктов
    for i, p in enumerate(products):
        p["id"] = i
    
    return {"message": f"Продукт '{deleted_product['name']}' удален"}

from typing import List, Optional
from pydantic import BaseModel, Field
 
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="Название продукта")
    price: float = Field(..., gt=0, description="Цена продукта в рублях")
    description: Optional[str] = Field(None, max_length=1000, description="Описание продукта")
 
class ProductCreate(ProductBase):
    pass
 
class ProductResponse(ProductBase):
    id: int = Field(..., description="Уникальный идентификатор продукта")
 
@app.post("/products/", response_model=ProductResponse)
def create_product(product: ProductCreate):
    new_product = product.dict()
    new_product["id"] = len(products)
    products.append(new_product)
    return new_product
 
@app.get("/products/{product_id}", response_model=ProductResponse)
def read_product(product_id: int = Path(..., ge=0)):
    if product_id >= len(products):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    return products[product_id]
 
@app.get("/products/", response_model=List[ProductResponse])
def list_products(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    min_price: float = Query(None, ge=0)
):
    result = products
    
    if min_price is not None:
        result = [p for p in result if p["price"] >= min_price]
    
    return result[skip : skip + limit]

app = FastAPI(
    title="API для управления продуктами",
    description="Этот API позволяет управлять каталогом продуктов",
    version="0.1.0",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Иван Иванов",
        "url": "http://example.com/contact/",
        "email": "ivan@example.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
)

from fastapi import Body
 
@app.post("/products/", response_model=ProductResponse)
def create_product(
    product: ProductCreate = Body(
        ...,
        example={
            "name": "Яблоко",
            "price": 1.5,
            "description": "Свежее красное яблоко"
        }
    )
):
    new_product = product.dict()
    new_product["id"] = len(products)
    products.append(new_product)
    return new_product

from pydantic import BaseModel, Field, validator
from typing import List, Optional
from datetime import date
 
class Category(BaseModel):
    name: str
    description: Optional[str] = None
 
class ProductDetailed(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    is_in_stock: bool = True
    tags: List[str] = []
    manufactured: date
    category: Category
    
    @validator('manufactured')
    def check_date(cls, v):
        if v > date.today():
            raise ValueError("Дата производства не может быть в будущем")
        return v
    
    @validator('tags')
    def check_tags(cls, v):
        if len(v) > 5:
            raise ValueError("Слишком много тегов (максимум 5)")
        return v

# Преобразование модели в JSON-строку
product_json = product.json()
 
# Преобразование из словаря в модель
product_data = {"name": "Молоко", "price": 60.5, "manufactured": "2023-05-01", 
                "category": {"name": "Молочные продукты"}}
product = ProductDetailed.parse_obj(product_data)

from fastapi import FastAPI, File, UploadFile
import shutil
from pathlib import Path
 
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)
    
    # Сохраняем загруженный файл
    file_path = upload_dir / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename, "size": file_path.stat().st_size}

@app.post("/upload-multiple/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        # Здесь может быть любая обработка файла
        results.append({"filename": file.filename, "content_type": file.content_type})
    return results

from fastapi.responses import FileResponse
 
@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = Path("uploads") / filename
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="Файл не найден")
        
    return FileResponse(file_path)

from fastapi import FastAPI, Header, Cookie
 
@app.get("/headers/")
async def read_headers(user_agent: str = Header(None)):
    return {"User-Agent": user_agent}
 
@app.get("/cookies/")
async def read_cookies(session: str = Cookie(None)):
    return {"session": session}

from fastapi.responses import JSONResponse
 
@app.post("/login/")
async def login(username: str):
    content = {"message": f"Привет, {username}"}
    response = JSONResponse(content=content)
    response.set_cookie(key="session", value="secret-token")
    return response

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
 
app = FastAPI()
 
class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name
 
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Упс! {exc.name} сломался. Мы уже чиним!"},
    )
 
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 42:
        raise CustomException(name="Продукт")
    return {"item_id": item_id}

from fastapi import FastAPI, Depends, HTTPException, Header
 
app = FastAPI()
 
async def verify_api_key(x_api_key: str = Header(None)):
    if x_api_key != "секретный_ключ":
        raise HTTPException(status_code=403, detail="Неверный API-ключ")
    return x_api_key
 
@app.get("/secure-endpoint/", dependencies=[Depends(verify_api_key)])
async def secure_endpoint():
    return {"message": "Это защищенный эндпоинт"}
 
@app.get("/another-secure/")
async def another_secure(api_key: str = Depends(verify_api_key)):
    return {"message": f"Ваш API-ключ: {api_key}"}

def get_current_user(token: str = Depends(oauth2_scheme)):
    # Проверка токена и возврат пользователя
    user = decode_token(token)
    if user is None:
        raise HTTPException(
            status_code=401, 
            detail="Недействительные учетные данные"
        )
    return user
 
def get_admin_user(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(
            status_code=403, 
            detail="Недостаточно прав"
        )
    return current_user
 
@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user
 
@app.get("/admin/stats")
def admin_stats(admin: User = Depends(get_admin_user)):
    return {"total_users": get_users_count()}

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 
# Хранилище для активных соединений
class ConnectionManager:
    def __init__(self):
        self.active_connections = []
 
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
 
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
 
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)
 
manager = ConnectionManager()
 
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # Эхо индивидуальному клиенту
            await websocket.send_text(f"Вы сказали: {data}")
            # Рассылка всем клиентам
            await manager.broadcast(f"Клиент #{client_id} говорит: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Клиент #{client_id} покинул чат")

import time
from fastapi import FastAPI, Request
 
app = FastAPI()
 
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
logger = logging.getLogger("api")
 
@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    try:
        response = await call_next(request)
        logger.info(f"Response: {response.status_code}")
        return response
    except Exception as e:
        logger.error(f"Error processing request: {str(e)}")
        return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

from fastapi.middleware.cors import CORSMiddleware
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost", "https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

import asyncio
from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/sync")
def sync_route():
    # Блокирующая операция - блокирует весь сервер
    import time
    time.sleep(1)
    return {"message": "Синхронный путь"}
 
@app.get("/async")
async def async_route():
    # Неблокирующая операция - другие запросы обрабатываются параллельно
    await asyncio.sleep(1)
    return {"message": "Асинхронный путь"}

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient
 
app = FastAPI()
 
@app.on_event("startup")
async def startup_db_client():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
    app.mongodb = app.mongodb_client.db
 
@app.on_event("shutdown")
async def shutdown_db_client():
    app.mongodb_client.close()
 
@app.get("/users/{user_id}")
async def get_user(user_id: str):
    user = await app.mongodb.users.find_one({"_id": user_id})
    if user:
        return user
    return {"error": "User not found"}

from fastapi import FastAPI, BackgroundTasks
 
app = FastAPI()
 
def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")
 
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Отправка уведомления на {email}")
    return {"message": "Уведомление будет отправлено в фоновом режиме"}

from typing import List
from pydantic import BaseModel
from fastapi import FastAPI
 
app = FastAPI()
 
class Product(BaseModel):
    name: str
    price: float
    quantity: int = 1
 
@app.post("/calculate-total/")
async def calculate_total(products: List[Product]):
    total = sum(product.price * product.quantity for product in products)
    return {"total": total}

from typing import Dict, Any, Union, Optional
from fastapi import FastAPI, Query, Path, Body
 
app = FastAPI()
 
@app.get("/items/{item_id}", response_model=Dict[str, Any])
async def read_item(
    item_id: int = Path(..., title="ID товара"),
    q: Optional[str] = Query(None, max_length=50),
    skip: int = Query(0, ge=0),
    limit: Union[int, None] = Query(10, le=100)
):
    # ...
    return {"item_id": item_id, "q": q, "skip": skip, "limit": limit}

class UserInDB(BaseModel):
    id: int
    username: str
    password_hash: str
    email: str
    is_admin: bool = False
 
class UserPublic(BaseModel):
    id: int
    username: str
    email: str
 
@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int):
    # Предположим, это реальный объект с хешем пароля и другими секретами
    user = UserInDB(
        id=user_id,
        username="johndoe",
        password_hash="supersecret",
        email="john@example.com",
        is_admin=True
    )
    return user  # FastAPI автоматически отфильтрует поля password_hash и is_admin

@app.get("/items/")
async def read_items():
    return [{"name": "Item 1"}, {"name": "Item 2"}]
 
@app.post("/items/")
async def create_item(item: dict):
    return item
 
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: dict):
    return {"item_id": item_id, **item}
 
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"deleted": item_id}
 
@app.patch("/items/{item_id}")
async def partial_update(item_id: int, item: dict):
    return {"item_id": item_id, **item}

from fastapi import APIRouter, FastAPI
 
app = FastAPI()
 
# Роутер для пользователей
user_router = APIRouter(prefix="/users", tags=["users"])
 
@user_router.get("/")
async def list_users():
    return [{"username": "rick"}, {"username": "morty"}]
 
@user_router.get("/{username}")
async def get_user(username: str):
    return {"username": username}
 
# Роутер для товаров
product_router = APIRouter(prefix="/products", tags=["products"])
 
@product_router.get("/")
async def list_products():
    return [{"name": "Hammer"}, {"name": "Wrench"}]
 
# Подключаем роутеры к основному приложению
app.include_router(user_router)
app.include_router(product_router)

def create_crud_router(model, db_dependency):
    router = APIRouter()
    
    @router.get("/")
    async def list_items(db = Depends(db_dependency)):
        return await db.find_all(model)
    
    @router.get("/{item_id}")
    async def get_item(item_id: str, db = Depends(db_dependency)):
        return await db.find_one(model, item_id)
    
    # ... другие CRUD-методы
    
    return router
 
# Использование
user_router = create_crud_router(User, get_db)
app.include_router(user_router, prefix="/users", tags=["users"])

from fastapi import FastAPI, Depends
from functools import lru_cache
 
app = FastAPI()
 
class Settings:
    def __init__(self):
        # В реальном проекте здесь могла бы быть загрузка из .env файла
        self.app_name = "Awesome API"
        self.admin_email = "admin@example.com"
        self.items_per_page = 20
 
@lru_cache()
def get_settings():
    return Settings()
 
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email
    }

import aioredis
from fastapi import FastAPI, Depends
import json
 
app = FastAPI()
 
async def get_redis():
    redis = await aioredis.create_redis_pool("redis://localhost")
    try:
        yield redis
    finally:
        redis.close()
        await redis.wait_closed()
 
@app.get("/products/{product_id}")
async def get_product(product_id: int, redis = Depends(get_redis)):
    # Пробуем получить из кеша
    cached = await redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)
    
    # Если нет в кеше, получаем из БД
    product = await db.get_product(product_id)
    
    # Сохраняем в кеш на 1 час
    await redis.set(
        f"product:{product_id}", 
        json.dumps(product), 
        expire=3600
    )
    
    return product

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, Float
 
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
 
Base = declarative_base()
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
class Product(Base):
__tablename__ = "products"
 
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String)
price = Column(Float)
 
app = FastAPI()
 
async def get_db():
db = AsyncSessionLocal()
try:
    yield db
finally:
    await db.close()
 
@app.get("/products/{product_id}")
async def get_product(product_id: int, db: AsyncSession = Depends(get_db)):
from sqlalchemy import select
query = select(Product).where(Product.id == product_id)
result = await db.execute(query)
product = result.scalars().first()
 
if not product:
    raise HTTPException(status_code=404, detail="Продукт не найден")
return product

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel
 
# Настройки безопасности
SECRET_KEY = "ваш_секретный_ключ"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = False
 
class Token(BaseModel):
access_token: str
token_type: str
 
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
 
def get_user(db, username: str):
# В реальности - запрос к БД
if username == "testuser":
    return {"username": "testuser", "hashed_password": pwd_context.hash("password")}
 
def authenticate_user(db, username: str, password: str):
user = get_user(db, username)
if not user:
    return False
if not verify_password(password, user["hashed_password"]):
    return False
return user
 
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})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
app = FastAPI()
 
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(None, form_data.username, form_data.password)
if not user:
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Неверное имя пользователя или пароль",
        headers={"WWW-Authenticate": "Bearer"},
    )
access_token = create_access_token(
    data={"sub": user["username"]},
    expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {"access_token": access_token, "token_type": "bearer"}
 
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
except JWTError:
    raise credentials_exception
user = get_user(None, username=username)
if user is None:
    raise credentials_exception
return user
 
@app.get("/users/me", response_model=User)
async def read_users_me(current_user = Depends(get_current_user)):
return current_user

my_app/
├── alembic/               # Миграции базы данных
├── app/
│   ├── __init__.py
│   ├── main.py            # Точка входа FastAPI
│   ├── config.py          # Конфигурация приложения
│   ├── dependencies.py    # Общие зависимости
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/            # API версии 1
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/ # Маршруты сгруппированы по ресурсам
│   │   │   │   ├── __init__.py
│   │   │   │   ├── users.py
│   │   │   │   ├── products.py
│   │   │   │   └── ...
│   │   │   └── router.py  # Сборка всех маршрутов v1
│   │   └── v2/            # API версии 2
│   ├── core/              # Ядро приложения
│   │   ├── __init__.py
│   │   ├── security.py    # JWT, аутентификация
│   │   └── errors.py      # Обработчики ошибок
│   ├── crud/              # Функции CRUD
│   │   ├── __init__.py
│   │   ├── base.py        # Базовый класс CRUD
│   │   ├── users.py
│   │   └── ...
│   ├── db/                # Настройки БД
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── session.py
│   ├── models/            # SQLAlchemy модели
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── ...
│   └── schemas/           # Pydantic модели
│       ├── __init__.py
│       ├── user.py
│       └── ...
├── tests/                 # Тесты
└── .env                   # Переменные окружения

from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import JSONResponse
import aioredis
import aiohttp
from arq import create_pool
from arq.connections import RedisSettings
 
app = FastAPI()
 
async def process_video(ctx, video_id: str):
# Долгая обработка видео
return {"processed": True, "video_id": video_id}
 
@app.on_event("startup")
async def startup():
app.redis = await create_pool(RedisSettings(host="localhost"))
 
@app.on_event("shutdown")
async def shutdown():
await app.redis.close()
 
@app.post("/videos/{video_id}/process")
async def start_video_processing(video_id: str):
# Добавляем задачу в очередь
job = await app.redis.enqueue_job("process_video", video_id)
return {"job_id": job.job_id, "status": "processing"}
 
@app.get("/jobs/{job_id}")
async def get_job_status(job_id: str):
job = await app.redis.get_job_result(job_id)
if job is None:
    return {"status": "processing"}
return {"status": "completed", "result": job}

from fastapi import FastAPI, Request, Response, Depends
from fastapi.responses import JSONResponse
import time
import hashlib
 
app = FastAPI()
 
def generate_etag(data: dict) -> str:
    # Создаем хеш содержимого для ETag
    content = str(sorted(data.items())).encode()
    return hashlib.md5(content).hexdigest()
 
@app.middleware("http")
async def add_cache_headers(request: Request, call_next):
    response = await call_next(request)
    
    # Добавляем заголовки кеширования только для GET-запросов
    if request.method == "GET":
        response.headers["Cache-Control"] = "max-age=60"  # Кешируем на 1 минуту
        
        # Если есть тело ответа, добавляем ETag
        if hasattr(response, "body"):
            try:
                body = response.body.decode()
                import json
                data = json.loads(body)
                etag = generate_etag(data)
                response.headers["ETag"] = etag
            except:
                pass
    
    return response

@app.get("/users/")
async def list_users(cursor: str = None, limit: int = 20):
    if cursor:
        # Декодируем курсор (обычно это закодированный timestamp последнего элемента)
        import base64
        last_timestamp = float(base64.b64decode(cursor).decode())
        users = await db.fetch_users(
            where="created_at < :timestamp", 
            params={"timestamp": last_timestamp},
            limit=limit + 1,  # +1 чтобы узнать, есть ли еще элементы
            order_by="created_at DESC"
        )
    else:
        users = await db.fetch_users(limit=limit + 1, order_by="created_at DESC")
    
    has_more = len(users) > limit
    result = users[:limit]  # Отрезаем лишний элемент
    
    # Создаем новый курсор, если есть еще элементы
    next_cursor = None
    if has_more and result:
        last_user = result[-1]
        next_cursor = base64.b64encode(str(last_user["created_at"]).encode()).decode()
    
    return {
        "users": result,
        "has_more": has_more,
        "next_cursor": next_cursor
    }

from fastapi.middleware.gzip import GZipMiddleware
 
app.add_middleware(GZipMiddleware, minimum_size=1000)  # Сжимать ответы больше 1KB

from fastapi import FastAPI
from starlette_exporter import PrometheusMiddleware, handle_metrics
 
app = FastAPI()
 
# Добавляем middleware для сбора метрик
app.add_middleware(PrometheusMiddleware)
 
# Эндпоинт для Prometheus
app.add_route("/metrics", handle_metrics)

import logging
import json
from datetime import datetime
 
class JSONFormatter(logging.Formatter):
def format(self, record):
    log_record = {
        "timestamp": datetime.utcnow().isoformat(),
        "level": record.levelname,
        "message": record.getMessage(),
        "module": record.module,
        "function": record.funcName,
    }
    
    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)
 
# Настройка логгера
logger = logging.getLogger("api")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)
 
# Middleware для добавления request_id в логи
@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    
    # Добавляем request_id в контекст логгера
    logger_with_context = logging.LoggerAdapter(
        logger, {"request_id": request_id}
    )
    
    # Используем этот логгер в обработчике
    request.state.logger = logger_with_context
    
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

from fastapi import FastAPI, Depends, HTTPException, Query, Path
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
from pydantic import BaseModel, Field
import uvicorn
 
app = FastAPI(title="Простой API каталога продуктов")
 
# Middleware для CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
# Модели данных
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    description: Optional[str] = None
 
class ProductCreate(ProductBase):
    pass
 
class ProductResponse(ProductBase):
    id: int
 
# Имитация базы данных
products_db = []
 
# Маршруты API
@app.post("/products/", response_model=ProductResponse)
def create_product(product: ProductCreate):
    new_product = product.dict()
    new_product["id"] = len(products_db)
    products_db.append(new_product)
    return new_product
 
@app.get("/products/{product_id}", response_model=ProductResponse)
def read_product(product_id: int = Path(..., ge=0)):
    if product_id >= len(products_db):
        raise HTTPException(status_code=404, detail="Продукт не найден")
    return products_db[product_id]
 
@app.get("/products/", response_model=List[ProductResponse])
def list_products(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    min_price: Optional[float] = Query(None, ge=0)
):
    result = products_db
    
    if min_price is not None:
        result = [p for p in result if p["price"] >= min_price]
    
    return result[skip : skip + limit]
 
if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)