Как ускорить разработку на FastAPI с помощью AI-промтов: проверенные шаблоны для эндпоинтов, валидации и фоновых задач
FastAPI — один из самых быстрорастущих фреймворков для создания API на Python. Его популярность объясняется высокой производительностью (сравнимой с Node.js и Go), встроенной поддержкой асинхронности и автоматической генерацией документации. По данным опроса JetBrains Python Developers Survey 2025, FastAPI используют 23% Python-разработчиков, и этот показатель продолжает расти. Однако даже опытные разработчики тратят до 40% времени на написание шаблонного кода: создание эндпоинтов, настройку Pydantic-схем и организацию фоновых задач.
AI-инструменты, такие как ChatGPT, Claude и Gemini, могут сократить это время в разы. Главное — правильно формулировать промты. В этой статье я собрал 5 проверенных шаблонов промтов, которые использую сам в ежедневной работе. Каждый промт — это не абстрактная идея, а конкретный запрос, который даёт готовый код. Все примеры актуальны для FastAPI 0.115.x и выше, Pydantic v2 и Python 3.12+.
Промт 1: Создание CRUD-эндпоинта с Pydantic-валидацией
Ситуация: Вам нужно быстро создать REST-эндпоинт для работы с сущностью (например, пользователями). Вместо того чтобы вручную писать модель SQLAlchemy, Pydantic-схемы и хендлеры, можно одним запросом получить готовый код.
Промт:
Напиши FastAPI-приложение с CRUD-эндпоинтами для модели User. Используй SQLAlchemy async (2.0+) и Pydantic v2 для валидации. Схемы: UserCreate (name: str, email: EmailStr, age: int > 0), UserResponse (id: int, name: str, email: str, age: int). Эндпоинты: POST /users (создание), GET /users/{id} (получение), GET /users (список с пагинацией через limit и offset), PUT /users/{id} (обновление), DELETE /users/{id} (удаление). Добавь обработку ошибок: 404 если пользователь не найден, 422 если валидация не пройдена. Используй зависимости (Depends) для сессии БД.
Результат:
from fastapi import FastAPI, HTTPException, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select, delete
from pydantic import BaseModel, EmailStr, field_validator
from typing import Optional
# --- Database setup ---
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
class UserModel(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
name: Mapped[str]
email: Mapped[str] = mapped_column(unique=True, index=True)
age: Mapped[int]
# --- Pydantic schemas ---
class UserCreate(BaseModel):
name: str
email: EmailStr
age: int
@field_validator("age")
@classmethod
def check_age(cls, v):
if v <= 0:
raise ValueError("Age must be positive")
return v
class UserResponse(BaseModel):
id: int
name: str
email: str
age: int
model_config = {"from_attributes": True}
# --- Dependencies ---
async def get_db():
async with AsyncSessionLocal() as session:
yield session
# --- App ---
app = FastAPI()
@app.on_event("startup")
async def startup():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
db_user = UserModel(**user.model_dump())
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(UserModel).where(UserModel.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users", response_model=list[UserResponse])
async def list_users(limit: int = Query(10, ge=1), offset: int = Query(0, ge=0), db: AsyncSession = Depends(get_db)):
result = await db.execute(select(UserModel).offset(offset).limit(limit))
return result.scalars().all()
@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user: UserCreate, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(UserModel).where(UserModel.id == user_id))
db_user = result.scalar_one_or_none()
if not db_user:
raise HTTPException(status_code=404, detail="User not found")
for key, value in user.model_dump().items():
setattr(db_user, key, value)
await db.commit()
await db.refresh(db_user)
return db_user
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(delete(UserModel).where(UserModel.id == user_id))
if result.rowcount == 0:
raise HTTPException(status_code=404, detail="User not found")
await db.commit()
Этот код можно сразу скопировать и запустить. Замените DATABASE_URL на нужный (PostgreSQL, MySQL) — AI подберёт соответствующий драйвер, если уточнить в промте.
Промт 2: Сложная Pydantic-валидация с кастомными правилами
Ситуация: Вам нужно валидировать нестандартные данные: например, пароль с требованиями по длине и символам, или поле, которое зависит от другого поля.
Промт:
Создай Pydantic v2 модель для регистрации пользователя с полями: username (строка 3-20 символов, только буквы и цифры), password (минимум 8 символов, хотя бы одна заглавная буква, одна цифра и один спецсимвол), confirm_password (должен совпадать с password), email (валидный EmailStr), age (int, 13-120), agree_to_terms (bool, обязательно True). Напиши кастомный валидатор на уровне модели для проверки совпадения паролей и field_validator для username (проверка на занятость — заглушка). Используй model_validator для проверки after=True.
Результат:
from pydantic import BaseModel, EmailStr, field_validator, model_validator
from typing import Optional
import re
class UserRegistration(BaseModel):
username: str
password: str
confirm_password: str
email: EmailStr
age: int
agree_to_terms: bool
@field_validator("username")
@classmethod
def validate_username(cls, v):
if not re.match(r"^[a-zA-Z0-9]{3,20}$", v):
raise ValueError("Username must be 3-20 alphanumeric characters")
# Заглушка проверки на занятость
taken_usernames = {"admin", "root", "user"}
if v.lower() in taken_usernames:
raise ValueError("Username already taken")
return v
@field_validator("password")
@classmethod
def validate_password(cls, v):
if len(v) < 8:
raise ValueError("Password must be at least 8 characters")
if not re.search(r"[A-Z]", v):
raise ValueError("Password must contain at least one uppercase letter")
if not re.search(r"[0-9]", v):
raise ValueError("Password must contain at least one digit")
if not re.search(r"[!@#$%^&*(),.?\":{}|<>_+\-=\[\]\\;\'`~]", v):
raise ValueError("Password must contain at least one special character")
return v
@field_validator("age")
@classmethod
def validate_age(cls, v):
if v < 13 or v > 120:
raise ValueError("Age must be between 13 and 120")
return v
@field_validator("agree_to_terms")
@classmethod
def validate_terms(cls, v):
if not v:
raise ValueError("You must agree to the terms")
return v
@model_validator(mode="after")
def check_passwords_match(self):
if self.password != self.confirm_password:
raise ValueError("Passwords do not match")
return self
Этот промт особенно полезен, когда нужно быстро набросать сложную бизнес-логику валидации, не отвлекаясь на синтаксис Pydantic.
Промт 3: Фоновая обработка с Celery и FastAPI
Ситуация: Вам нужно добавить фоновую задачу, которая выполняется после ответа клиенту. Например, отправка email или генерация отчёта.
Промт:
Напиши FastAPI-приложение, которое принимает POST /report с полем report_type (строка) и запускает фоновую задачу через Celery. Celery-задача генерирует файл (имитация — time.sleep(10)) и сохраняет в папке /reports. Используй Redis как брокер. Добавь эндпоинт GET /report/{task_id} для проверки статуса задачи (Pending, Started, Success). Для хранения статуса используй Redis (запись по ключу task:{task_id}). Учти, что FastAPI и Celery должны быть в одном проекте. Настрой логирование.
Результат:
# app.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from celery import Celery
from pydantic import BaseModel
import time
import logging
import redis
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Celery setup
celery_app = Celery(
"tasks",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/0"
)
# Redis client for status tracking
r = redis.Redis(host="localhost", port=6379, db=1)
@celery_app.task(bind=True, max_retries=3)
def generate_report(self, report_type: str):
task_id = self.request.id
r.set(f"task:{task_id}:status", "Started")
logger.info(f"Task {task_id}: generating report type {report_type}")
try:
# Имитация длительной работы
time.sleep(10)
with open(f"/reports/{task_id}.txt", "w") as f:
f.write(f"Report type: {report_type}\nGenerated at: {time.time()}")
r.set(f"task:{task_id}:status", "Success")
r.set(f"task:{task_id}:result", f"/reports/{task_id}.txt")
except Exception as e:
logger.error(f"Task {task_id} failed: {e}")
r.set(f"task:{task_id}:status", "Failed")
raise self.retry(exc=e, countdown=5)
# FastAPI app
app = FastAPI()
class ReportRequest(BaseModel):
report_type: str
@app.post("/report")
async def create_report(request: ReportRequest):
task = generate_report.delay(request.report_type)
r.set(f"task:{task.id}:status", "Pending")
return {"task_id": task.id, "status": "Pending"}
@app.get("/report/{task_id}")
async def get_report_status(task_id: str):
status = r.get(f"task:{task_id}:status")
if not status:
raise HTTPException(status_code=404, detail="Task not found")
result = r.get(f"task:{task_id}:result")
return {
"task_id": task_id,
"status": status.decode(),
"result": result.decode() if result else None
}
Важно: для работы с Redis и Celery потребуется установить celery[redis] и запустить Redis-сервер. Промт даёт рабочий скелет, который можно расширять.
Промт 4: Асинхронные эндпоинты с внешними API (httpx)
Ситуация: Ваш эндпоинт должен сделать несколько параллельных запросов к внешним API и агрегировать результаты.
Промт:
Напиши FastAPI-эндпоинт GET /aggregate, который одновременно запрашивает данные с трёх публичных API: https://api.github.com/users/{username} (информация о пользователе GitHub), https://api.github.com/users/{username}/repos (список репозиториев), https://api.github.com/users/{username}/orgs (список организаций). Используй httpx.AsyncClient с asyncio.gather. Если один из запросов упадёт с ошибкой, верни None для этого поля, но не роняй весь эндпоинт. Обработай таймаут (5 секунд на каждый запрос). Верни JSON с полями: user, repos, orgs.
Результат:
from fastapi import FastAPI, HTTPException, Query
import httpx
import asyncio
from typing import Optional, Any
app = FastAPI()
TIMEOUT = 5.0
async def fetch_data(client: httpx.AsyncClient, url: str) -> Optional[Any]:
try:
response = await client.get(url, timeout=TIMEOUT)
response.raise_for_status()
return response.json()
except Exception as e:
return None
@app.get("/aggregate")
async def aggregate(username: str = Query(..., min_length=1)):
async with httpx.AsyncClient() as client:
user_url = f"https://api.github.com/users/{username}"
repos_url = f"https://api.github.com/users/{username}/repos"
orgs_url = f"https://api.github.com/users/{username}/orgs"
user_data, repos_data, orgs_data = await asyncio.gather(
fetch_data(client, user_url),
fetch_data(client, repos_url),
fetch_data(client, orgs_url)
)
if user_data is None:
raise HTTPException(status_code=404, detail="User not found or API error")
return {
"user": user_data,
"repos": repos_data,
"orgs": orgs_data
}
Этот промт полезен для микросервисной архитектуры, где нужно агрегировать данные из нескольких источников.
Промт 5: Генерация Swagger-документации с описаниями
Ситуация: FastAPI автоматически генерирует OpenAPI-документацию, но описания эндпоинтов и полей по умолчанию пустые. Хочется получить красивую доку с примерами.
Промт:
Дополни следующий FastAPI-код, добавив в Pydantic-схемы и эндпоинты подробные описания (title, description, examples) для документации. Используй Field для каждого поля и docstring для хендлеров. Включи примеры запросов и ответов. Для эндпоинта POST /users добавь пример тела запроса и ответа. Для GET /users/{id} добавь описание ошибки 404. Используй response_description и summary.
Пример кода для промта:
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import Optional
app = FastAPI(title="User API", description="API for managing users", version="1.0.0")
class UserCreate(BaseModel):
name: str = Field(..., title="Full name", description="User's full name", example="John Doe")
email: str = Field(..., title="Email address", description="Valid email", example="john@example.com")
age: int = Field(..., title="Age", ge=0, le=150, description="User's age", example=30)
class UserResponse(BaseModel):
id: int = Field(..., title="User ID", description="Unique identifier", example=1)
name: str = Field(..., title="Full name", example="John Doe")
email: str = Field(..., title="Email address", example="john@example.com")
age: int = Field(..., title="Age", example=30)
@app.post(
"/users",
response_model=UserResponse,
summary="Create a new user",
response_description="The created user",
status_code=201
)
async def create_user(user: UserCreate):
"""
Create a new user with the provided data.
- **name**: User's full name
- **email**: Must be a valid email format
- **age**: Must be between 0 and 150
"""
# Здесь логика создания
return UserResponse(id=1, **user.model_dump())
@app.get(
"/users/{user_id}",
response_model=UserResponse,
summary="Get user by ID",
responses={
404: {"description": "User not found"}
}
)
async def get_user(user_id: int):
"""
Retrieve a user by their unique ID.
Returns 404 if the user does not exist.
"""
# Здесь логика получения
pass
После добавления описаний документация в /docs станет значительно информативнее — с примерами, понятными описаниями полей и кодов ответов.
Заключение
FastAPI — мощный инструмент, но даже с ним много времени уходит на рутинные операции. Промты помогают сосредоточиться на бизнес-логике, а не на синтаксисе. Все приведённые шаблоны можно адаптировать под конкретные задачи: заменить базу данных, добавить аутентификацию, интегрировать с внешними сервисами.
На практике я использую эти промты каждый день. Они экономят от 10 до 30 минут на каждой задаче. Попробуйте — и вы увидите, как AI может стать вашим партнёром в коде, а не просто игрушкой.
Комментарии