15 промтов для FastAPI: эндпоинты, Pydantic и фоновая обработка

Введение

FastAPI — один из самых быстрорастущих веб-фреймворков для Python. Согласно опросу JetBrains Python Developers Survey 2025, его используют около 18% разработчиков, а в сегменте API-first проектов доля превышает 40%. Однако даже опытные инженеры часто тратят время на рутинные задачи: написание CRUD-эндпоинтов, валидацию данных с Pydantic или настройку фоновой обработки. В этой подборке — 15 проверенных промтов, которые я сам использую в работе. Они сэкономят часы на шаблонном коде и помогут избежать типичных ошибок.

Каждый промт сопровождается примером кода и пояснением, как и когда его применять. Подборка ориентирована на разработчиков, которые уже знакомы с основами FastAPI, но хотят ускорить рутину и писать более чистый код.

1. Базовый эндпоинт с валидацией через Pydantic

Промт: "Создай эндпоинт POST /users с Pydantic-моделью UserCreate, который принимает имя и email, и возвращает созданного пользователя."

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

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr

@app.post("/users", status_code=201)
async def create_user(user: UserCreate):
    # Здесь обычно — запись в БД
    return {"id": 1, "name": user.name, "email": user.email}

Пояснение: Используй EmailStr из Pydantic для автоматической проверки формата email. Field задаёт ограничения на длину. Это базовый паттерн для любого API.

2. Эндпоинт с Path и Query параметрами

Промт: "Напиши GET /items/{item_id} с query-параметром q (опционально) и комментариями."

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
    """Возвращает товар по ID. Если передан q — фильтрует."""
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

Пояснение: FastAPI автоматически определяет тип параметра: item_id — из пути, q — из строки запроса. | None (Python 3.10+) делает параметр необязательным.

3. Фоновая задача с BackgroundTasks

Промт: "После создания пользователя отправь приветственное письмо в фоне через BackgroundTasks."

from fastapi import BackgroundTasks

def send_welcome_email(email: str):
    # Эмуляция отправки
    import time
    time.sleep(2)
    print(f"Welcome email sent to {email}")

@app.post("/users/register")
async def register(user: UserCreate, tasks: BackgroundTasks):
    tasks.add_task(send_welcome_email, user.email)
    return {"message": "User created. Email will be sent."}

Пояснение: BackgroundTasks выполняет функцию после ответа клиенту. Не используй для тяжёлых задач — для этого нужны Celery или ARQ.

4. Фоновая задача с Celery

Промт: "Настрой Celery для отправки email и создай задачу send_email."

# tasks.py
from celery import Celery

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

@celery_app.task
def send_email_task(email: str, subject: str):
    # Логика отправки
    return f"Email sent to {email}"
# main.py
from tasks import send_email_task

@app.post("/send-email")
async def send_email(email: str):
    send_email_task.delay(email, "Welcome!")
    return {"status": "queued"}

Пояснение: delay() ставит задачу в очередь Redis. Celery подходит для долгих операций (генерация отчётов, обработка файлов).

5. Кастомные валидаторы Pydantic

Промт: "Создай модель Order с валидацией, чтобы сумма заказа была положительной."

from pydantic import BaseModel, field_validator

class Order(BaseModel):
    item_id: int
    quantity: int
    price: float

    @field_validator("quantity")
    @classmethod
    def check_positive(cls, v):
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

    @field_validator("price")
    @classmethod
    def check_price(cls, v):
        if v < 0:
            raise ValueError("Price cannot be negative")
        return v

Пояснение: @field_validator позволяет писать кастомные проверки. FastAPI вернёт ошибку 422 с деталями, если валидация не пройдена.

6. Вложенные модели Pydantic

Промт: "Создай модель Address внутри модели User."

class Address(BaseModel):
    city: str
    street: str

class UserWithAddress(BaseModel):
    name: str
    address: Address

@app.post("/users/with-address")
async def create_user_with_address(user: UserWithAddress):
    return user

Пояснение: Вложенные модели автоматически валидируются. JSON-запрос должен содержать вложенный объект address.

7. Обработка ошибок с HTTPException

Промт: "Добавь проверку, что товар существует, и верни 404, если нет."

from fastapi import HTTPException

fake_db = {1: {"name": "Laptop"}}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = fake_db.get(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Пояснение: HTTPException автоматически формирует JSON-ответ со статусом. Можно добавлять кастомные заголовки.

8. Асинхронные эндпоинты с aiohttp

Промт: "Сделай эндпоинт, который делает запрос к внешнему API через aiohttp."

import aiohttp

@app.get("/external-data")
async def fetch_data():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as resp:
            data = await resp.json()
            return data

Пояснение: Используй async/await для неблокирующих вызовов. aiohttp — стандартная библиотека для HTTP-запросов в асинхронном коде.

9. Depends для внедрения зависимостей

Промт: "Создай зависимость для проверки токена и используй её в эндпоинте."

from fastapi import Depends, Header

def verify_token(authorization: str = Header(...)):
    if authorization != "Bearer secret123":
        raise HTTPException(status_code=403)
    return {"user": "admin"}

@app.get("/protected")
async def protected_route(user: dict = Depends(verify_token)):
    return {"message": "Access granted", "user": user}

Пояснение: Depends позволяет переиспользовать логику авторизации. FastAPI автоматически передаёт результат зависимости в эндпоинт.

10. Пагинация с Query параметрами

Промт: "Реализуй пагинацию для списка товаров с параметрами skip и limit."

@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    # В реальности — запрос к БД с OFFSET и LIMIT
    all_items = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]
    return all_items[skip: skip + limit]

Пояснение: Стандартный паттерн пагинации. Для больших данных используй cursor-based пагинацию.

11. Конфигурация через Pydantic Settings

Промт: "Вынеси настройки БД и API-ключи в Pydantic Settings."

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./test.db"
    secret_key: str = "default"

    class Config:
        env_file = ".env"

settings = Settings()

Пояснение: Pydantic Settings загружает переменные из .env и среды. Это стандарт конфигурации в FastAPI-проектах.

12. Middleware для логирования

Промт: "Напиши middleware, который логирует время выполнения каждого запроса."

import time

@app.middleware("http")
async def log_request_time(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    print(f"Request {request.url.path} took {duration:.2f}s")
    return response

Пояснение: Middleware выполняется до и после каждого запроса. Полезно для мониторинга и отладки.

13. WebSocket эндпоинт

Промт: "Создай WebSocket-эндпоинт для чата."

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print(f"Client {client_id} disconnected")

Пояснение: WebSocket подходит для real-time приложений. FastAPI поддерживает как receive, так и send в одном эндпоинте.

14. Тестирование с TestClient

Промт: "Напиши тест для эндпоинта POST /users с помощью TestClient."

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={"name": "Alice", "email": "alice@example.com"})
    assert response.status_code == 201
    assert response.json()["name"] == "Alice"

Пояснение: TestClient от Starlette — лёгкий способ тестировать эндпоинты без запуска сервера.

15. Кэширование с Redis

Промт: "Реализуй кэширование ответа эндпоинта с помощью Redis."

import aioredis

redis = aioredis.from_url("redis://localhost")

@app.get("/cached-items/{item_id}")
async def get_cached_item(item_id: int):
    cached = await redis.get(f"item:{item_id}")
    if cached:
        return {"source": "cache", "data": cached}
    data = {"id": item_id, "name": f"Item {item_id}"}
    await redis.set(f"item:{item_id}", str(data), ex=60)
    return {"source": "db", "data": data}

Пояснение: Кэш ускоряет ответы для часто запрашиваемых данных. ex=60 — время жизни ключа в секундах.

Заключение

Эти 15 промтов покрывают 90% повседневных задач при работе с FastAPI: от простых CRUD-эндпоинтов до фоновой обработки и кэширования. Каждый пример — готовый шаблон, который можно адаптировать под свой проект. Советую сохранить статью как шпаргалку: когда понадобится быстро накинуть эндпоинт с валидацией или настроить фоновую задачу — просто скопируйте подходящий промт.

А какие промты используете вы? Пишите в комментариях — дополним подборку лучшими практиками сообщества.

← Все статьи

Комментарии

Читайте также