FastAPI для Python: как построить современный веб-сервис

Раздел: Веб-разработка -> Фреймворки (FastAPI)

FastAPI представляет собой современный веб-фреймворк для Python, ориентированный на создание высокопроизводительных API. Основанный на Starlette и Pydantic, он обеспечивает автоматическую валидацию данных, генерацию интерактивной документации OpenAPI и асинхронную обработку запросов. Фреймворк подходит для построения REST API, микросервисов, прототипов и производственных решений.

Основы работы с FastAPI

Наиболее эффективный способ начать работу с FastAPI - создать минимальное приложение с одним маршрутом и запустить его через Uvicorn.

Установка зависимостей:

pip install fastapi uvicorn

Fastapi веб разработка на python (веб-разработка на fastapi)

Пример файла main.py:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello World"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Пояснение: импортируется FastAPI, создаётся экземпляр приложения. Декоратор @app.get определяет обработчик GET-запроса к корневому пути. Функция возвращает словарь, который FastAPI автоматически преобразует в JSON. Запуск через Uvicorn с указанием хоста и порта.

Типичная ошибка при запуске

Если Uvicorn не установлен или команда запуска выполняется не из того окружения, возникает ошибка импорта. Решение: активировать виртуальное окружение и выполнить установку ещё раз.

Другая распространённая ошибка - отсутствие блока if __name__ == "__main__". При запуске скрипта напрямую без этой конструкции приложение не стартует.

Как обрабатывать параметры пути и строки запроса?

FastAPI позволяет получать параметры из пути URL и query-строки с автоматической проверкой типов.

from fastapi import FastAPI, Path, Query

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(..., title="ID товара"),
    q: str | None = Query(None, max_length=50, description="Поисковый запрос")
):
    return {"item_id": item_id, "q": q}

Пояснение: параметр item_id берётся из пути и объявляется как целое число с помощью Path. q - необязательный query-параметр с ограничением длины. FastAPI автоматически генерирует документацию и проверяет типы при запросе.

Проблема: неверный тип данных

Если клиент передаёт строку в item_id (например, /items/abc), FastAPI вернёт ошибку 422 Unprocessable Entity. Решение: всегда указывать корректные типы в сигнатуре функции.

Как принимать данные POST через тело запроса?

Для обработки POST запросов с JSON-телом используется модель Pydantic.

from pydantic import BaseModel
from fastapi import FastAPI

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool | None = None

app = FastAPI()

@app.post("/items/")
def create_item(item: Item):
    item_dict = item.dict()
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

Пояснение: класс Item наследуется от BaseModel - его поля автоматически валидируются. FastAPI извлекает JSON из тела запроса, создаёт экземпляр модели и передаёт в функцию. Ответ также будет преобразован в JSON.

Ошибка сериализации

Если не указать response_model в декораторе, FastAPI может попытаться вернуть сложный объект, который не удаётся сериализовать. Рекомендуется явно указывать response_model=Item или возвращать dict.

Как организовать приложение с роутерами?

При росте проекта маршруты можно группировать с помощью APIRouter и подключать их к основному приложению.

# app/items.py
from fastapi import APIRouter

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/")
async def list_items():
    return [{"name": "Foo"}, {"name": "Bar"}]

@router.get("/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

# main.py
from fastapi import FastAPI
from app.items import router

app = FastAPI()
app.include_router(router)

Пояснение: в отдельном файле создаётся роутер с общим префиксом и тегом. В главном файле роутер подключается через include_router. Это упрощает поддержку и масштабирование.

Конфликт префиксов

Если в роутере уже задан префикс, а при подключении указать другой префикс, маршруты могут дублироваться. Следует задавать префикс только в одном месте - либо в роутере, либо в include_router.

Как подключить базу данных с SQLAlchemy и асинхронностью?

FastAPI хорошо работает с асинхронными библиотеками, например SQLAlchemy async. Пример настройки сессии базы данных:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# В маршруте
from fastapi import Depends, FastAPI

app = FastAPI()

@app.get("/users/")
async def get_users(db: AsyncSession = Depends(get_db)):
    # выполнение запроса
    result = await db.execute(...)
    return result.scalars().all()

Пояснение: создаётся асинхронный движок и фабрика сессий. Зависимость get_db предоставляет сессию для каждого запроса, обеспечивая изоляцию.

Ошибка при отсутствии драйвера

Для PostgreSQL требуется установить asyncpg. Если драйвер не найден, SQLAlchemy выдаст исключение. Решение: pip install asyncpg.

Как добавить поддержку CORS для фронтенда?

Чтобы разрешить запросы с других доменов, используется CORSMiddleware.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Пояснение: middleware добавляется до определения маршрутов. Параметр allow_origins - список разрешённых источников. Для разработки можно указать ["*"], но для продакшена стоит задать конкретные домены.

Политика CORS блокирует запросы

Если фронтенд отправляет запрос с неразрешённого источника, браузер блокирует ответ. Необходимо проверить список allow_origins и при необходимости добавить нужный домен или использовать символ * для всех.

Как реализовать аутентификацию через JWT?

FastAPI предоставляет инструменты OAuth2 для работы с токенами. Пример с JWT:

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta

SECRET_KEY = "secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    email: str | None = None

# Вспомогательные функции
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # проверка пользователя (заглушка)
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    token = create_access_token(data={"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return User(**user)

Пояснение: используется OAuth2PasswordBearer для получения токена из заголовка Authorization. Функция create_access_token формирует JWT с временем жизни. В маршруте извлекается и проверяется токен.

Ошибка декодирования JWT

Если секретный ключ не совпадает или токен просрочен, возникает JWTError. Следует обрабатывать это исключение и возвращать 401 статус.

Как использовать Dependency Injection для общих зависимостей?

Зависимости (Dependencies) позволяют вынести общую логику, например, подключение к БД или проверку прав.

from fastapi import Depends, FastAPI, Header, HTTPException

async def verify_token(x_token: str = Header()):
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

app = FastAPI()

@app.get("/secure-data")
async def get_secure_data(token: str = Depends(verify_token)):
    return {"data": "confidential", "token": token}

Пояснение: функция verify_token объявляет параметр заголовка, проверяет его значение и возвращает токен. В маршруте эта функция вызывается через Depends, результат передаётся в обработчик.

Некорректная передача зависимости

Если зависимость не объявлена как аргумент с Depends, FastAPI не распознает её как зависимость и передаст значение из тела запроса.

В этом разделе представлены расширенные примеры использования FastAPI, которые демонстрируют реальные сценарии разработки.

Расширенные практические примеры с FastAPI

1. Полный CRUD с SQLAlchemy async и PostgreSQL

Создаётся приложение для управления товарами с асинхронным доступом к базе данных.

Пример 
# models.py
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Float)
    description = Column(String, nullable=True)

# schemas.py
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    price: float
    description: str | None = None

class ItemResponse(ItemCreate):
    id: int
    class Config:
        from_attributes = True

# crud.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from . import models, schemas

async def get_items(db: AsyncSession, skip: int = 0, limit: int = 100):
    result = await db.execute(select(models.Item).offset(skip).limit(limit))
    return result.scalars().all()

async def create_item(db: AsyncSession, item: schemas.ItemCreate):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from .database import get_db, engine, Base
from . import crud, schemas

app = FastAPI()

@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.get("/items/", response_model=list[schemas.ItemResponse])
async def read_items(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    items = await crud.get_items(db, skip=skip, limit=limit)
    return items

@app.post("/items/", response_model=schemas.ItemResponse)
async def create_item(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    return await crud.create_item(db, item)
# Запрос
curl -X POST "http://localhost:8000/items/" -H "Content-Type: application/json" -d '{"name":"Widget","price":9.99}'
# Ответ
{"name":"Widget","price":9.99,"description":null,"id":1}

Пояснение: модели SQLAlchemy описывают таблицу, Pydantic схемы - валидацию входящих и исходящих данных. В CRUD-функциях выполняются асинхронные запросы. Маршруты используют зависимости для получения сессии базы данных.

2. WebSocket эхо-сервер

WebSocket позволяет установить двустороннюю связь между клиентом и сервером.

Пример 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")
# Клиент (JavaScript)
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => console.log(event.data);
ws.send("Hello");  // Получит "Echo: Hello"

Пояснение: соединение принимается, затем сервер ожидает текстовые сообщения и отправляет их обратно с префиксом. Исключение WebSocketDisconnect обрабатывается для корректного завершения.

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

Используется для выполнения действий после отправки ответа клиенту, например, отправка email.

Пример 
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/send-notification")
async def send_notification(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Task scheduled"}
# Запрос
curl -X POST "http://localhost:8000/send-notification?email=test@example.com"
# Ответ
{"message":"Task scheduled"}
# После ответа в файл log.txt запишется "Notification sent to test@example.com"

Пояснение: функция write_log выполняется асинхронно после возврата ответа. BackgroundTasks удобны для неблокирующих операций.

4. Пагинация с фильтрацией

Реализация постраничного вывода с поиском через query-параметры.

Пример 
from fastapi import FastAPI, Query

app = FastAPI()

fake_db = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]

@app.get("/items")
async def list_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: str | None = Query(None, min_length=2)
):
    result = fake_db
    if search:
        result = [item for item in result if search.lower() in item["name"].lower()]
    return result[skip: skip + limit]
curl "http://localhost:8000/items?skip=20&limit=5&search=Item%202"
[{"id":22,"name":"Item 22"},{"id":23,"name":"Item 23"}]

Пояснение: параметры skip и limit управляют смещением и количеством записей. Параметр search фильтрует по вхождению строки. FastAPI валидирует ограничения значений.

5. Middleware для логирования времени обработки

Пользовательское промежуточное ПО измеряет длительность выполнения запроса.

Пример 
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

app = FastAPI()
app.add_middleware(TimingMiddleware)

@app.get("/slow")
async def slow_endpoint():
    await asyncio.sleep(1)
    return {"message": "Done"}
curl -I "http://localhost:8000/slow"
HTTP/1.1 200 OK
x-process-time: 1.0034

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

6. Кастомная обработка ошибок с пользовательским ответом

Переопределение глобальных обработчиков исключений для возврата единообразного формата ошибок.

Пример 
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(CustomException)
async def custom_handler(request: Request, exc: CustomException):
    return JSONResponse(status_code=418, content={"error": f"Custom: {exc.name}"})

@app.get("/custom-error")
async def raise_custom():
    raise CustomException(name="Something broke")

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail, "code": exc.status_code})
curl "http://localhost:8000/custom-error"
{"error":"Custom: Something broke"}

curl "http://localhost:8000/nonexistent"
{"detail":"Not Found","code":404}

Пояснение: определяются пользовательские классы исключений и обработчики через декоратор @app.exception_handler. Это позволяет централизованно управлять форматом ответов при ошибках.

Веб-разработка на FastAPI - comments

En
Fastapi веб разработка на python (python)