FastAPI – асинхронный фреймворк для API

Что такое FastAPI

FastAPI — это современный высокопроизводительный веб-фреймворк для создания API на языке Python. Разработанный Себастьяном Рамиресом, этот фреймворк построен на основе стандартов OpenAPI и JSON Schema, что обеспечивает автоматическую генерацию документации и типизацию данных.

FastAPI использует новейшие возможности Python, включая аннотации типов (type hints) и асинхронное программирование, что делает его одним из самых быстрых Python-фреймворков для разработки API. По производительности он сравним с Node.js и Go, что делает его отличным выбором для высоконагруженных приложений.

Ключевые особенности FastAPI

Высокая производительность

FastAPI построен на основе Starlette для веб-части и Pydantic для валидации данных. Это обеспечивает производительность на уровне самых быстрых Python-фреймворков. Фреймворк полностью поддерживает ASGI (Asynchronous Server Gateway Interface), что позволяет обрабатывать тысячи одновременных подключений.

Автоматическая генерация документации

Одной из главных особенностей FastAPI является автоматическая генерация интерактивной документации API. Фреймворк создает документацию в формате OpenAPI (Swagger) и предоставляет два интерфейса:

• Swagger UI — доступен по адресу /docs
• ReDoc — доступен по адресу /redoc

Полная поддержка типизации

FastAPI использует стандартные аннотации типов Python для валидации данных, сериализации и автоматической генерации документации. Это обеспечивает отличную поддержку автодополнения в IDE и снижает количество ошибок во время разработки.

Встроенная валидация данных

Благодаря интеграции с Pydantic, FastAPI автоматически валидирует входящие данные, конвертирует типы и возвращает понятные сообщения об ошибках. Это значительно упрощает разработку и повышает надежность приложений.

Асинхронность из коробки

FastAPI полностью поддерживает асинхронное программирование с использованием async/await. Это позволяет создавать высокопроизводительные приложения, способные обрабатывать множество одновременных запросов.

Установка и базовая настройка

Установка FastAPI

Для установки FastAPI и всех необходимых зависимостей используйте следующую команду:

pip install fastapi[all]

Эта команда установит FastAPI вместе с Uvicorn (ASGI сервер) и другими полезными зависимостями.

Для минимальной установки используйте:

pip install fastapi
pip install uvicorn[standard]

Создание первого приложения

Создайте файл main.py с базовым приложением:

from fastapi import FastAPI

app = FastAPI(
    title="Мой FastAPI проект",
    description="Это пример FastAPI приложения",
    version="1.0.0"
)

@app.get("/")
async def root():
    return {"message": "Добро пожаловать в FastAPI!"}

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Запуск приложения

Запустите сервер разработки с помощью Uvicorn:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Параметры запуска:

• --reload — автоматическая перезагрузка при изменении кода
• --host 0.0.0.0 — принимать подключения от всех IP-адресов
• --port 8000 — порт для запуска сервера

Работа с HTTP-методами

FastAPI поддерживает все стандартные HTTP-методы через соответствующие декораторы:

GET-запросы

@app.get("/items/{item_id}")
async def get_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

@app.get("/items/")
async def get_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

POST-запросы

from pydantic import BaseModel

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

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item, "message": "Товар создан"}

PUT и PATCH-запросы

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

@app.patch("/items/{item_id}")
async def update_item_partial(item_id: int, item: dict):
    return {"item_id": item_id, "updated_fields": item}

DELETE-запросы

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": f"Товар {item_id} удален"}

Работа с параметрами запроса

Path-параметры

from fastapi import Path

@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(..., title="ID товара", ge=1)
):
    return {"item_id": item_id}

Query-параметры

from fastapi import Query

@app.get("/items/")
async def get_items(
    q: str = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return {"q": q, "skip": skip, "limit": limit}

Работа с заголовками

from fastapi import Header

@app.get("/items/")
async def get_items(
    user_agent: str = Header(None),
    x_token: str = Header(None, alias="X-Token")
):
    return {"User-Agent": user_agent, "X-Token": x_token}

Работа с cookies

from fastapi import Cookie

@app.get("/items/")
async def get_items(ads_id: str = Cookie(None)):
    return {"ads_id": ads_id}

Модели данных с Pydantic

Создание базовых моделей

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class User(BaseModel):
    id: Optional[int] = None
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., ge=0, le=150)
    created_at: datetime = Field(default_factory=datetime.now)
    
    class Config:
        schema_extra = {
            "example": {
                "name": "Иван Иванов",
                "email": "ivan@example.com",
                "age": 30
            }
        }

Вложенные модели

from typing import List

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

class UserWithAddress(BaseModel):
    name: str
    email: str
    addresses: List[Address]

Модели ответов

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

@app.post("/users/", response_model=UserResponse)
async def create_user(user: User):
    # Логика создания пользователя
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )

Асинхронное программирование

Асинхронные обработчики

import asyncio
import aiohttp

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

@app.get("/delayed")
async def delayed_response():
    await asyncio.sleep(2)
    return {"message": "Ответ через 2 секунды"}

Работа с асинхронными базами данных

import asyncpg
from databases import Database

DATABASE_URL = "postgresql://user:password@localhost/database"
database = Database(DATABASE_URL)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = "SELECT * FROM users WHERE id = :user_id"
    user = await database.fetch_one(query, {"user_id": user_id})
    return user

Обработка ошибок

Стандартные HTTP-исключения

from fastapi import HTTPException, status

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Товар не найден"
        )
    return {"item_id": item_id}

Пользовательские обработчики ошибок

from fastapi import Request
from fastapi.responses import JSONResponse

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}"}
    )

Зависимости и внедрение зависимостей

Простые зависимости

from fastapi import Depends

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def get_items(commons: dict = Depends(common_parameters)):
    return commons

Зависимости с состоянием

class DatabaseManager:
    def __init__(self):
        self.connection = None
    
    async def connect(self):
        # Подключение к базе данных
        pass
    
    async def disconnect(self):
        # Отключение от базы данных
        pass

db_manager = DatabaseManager()

async def get_db():
    return db_manager

@app.get("/items/")
async def get_items(db: DatabaseManager = Depends(get_db)):
    return {"database": "connected"}

Аутентификация и авторизация

OAuth2 с JWT

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

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, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        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
    return username

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # Проверка пользователя
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}"}

Middleware и CORS

Добавление CORS

from fastapi.middleware.cors import CORSMiddleware

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

Пользовательские middleware

from fastapi import Request
import time

@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

Работа с базами данных

SQLAlchemy с FastAPI

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
async def create_user(user: User, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Структура проекта

Рекомендуемая структура для больших проектов

project/
├── main.py                 # Точка входа
├── core/
│   ├── __init__.py
│   ├── config.py          # Конфигурация
│   ├── security.py        # Безопасность
│   └── database.py        # Подключение к БД
├── models/
│   ├── __init__.py
│   ├── user.py           # Модели пользователей
│   └── item.py           # Модели товаров
├── schemas/
│   ├── __init__.py
│   ├── user.py           # Pydantic схемы
│   └── item.py
├── api/
│   ├── __init__.py
│   ├── deps.py           # Зависимости
│   └── v1/
│       ├── __init__.py
│       ├── endpoints/
│       │   ├── __init__.py
│       │   ├── users.py
│       │   └── items.py
│       └── api.py
├── tests/
│   ├── __init__.py
│   └── test_main.py
└── requirements.txt

Использование APIRouter

from fastapi import APIRouter

router = APIRouter(
    prefix="/api/v1",
    tags=["items"],
    responses={404: {"description": "Not found"}}
)

@router.get("/items/")
async def get_items():
    return [{"item_id": 1}]

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

# В main.py
app.include_router(router)

Тестирование FastAPI приложений

Базовое тестирование

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Добро пожаловать в FastAPI!"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.5, "is_offer": True}
    )
    assert response.status_code == 200
    assert response.json()["item"]["name"] == "Test Item"

@pytest.mark.asyncio
async def test_async_endpoint():
    response = client.get("/async-data")
    assert response.status_code == 200

Тестирование с базой данных

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

Развертывание FastAPI приложений

Использование Docker

FROM python:3.9

WORKDIR /app

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

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Продакшн с Gunicorn

pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Настройка Nginx

server {
    listen 80;
    server_name yourdomain.com;
    
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

WebSocket поддержка

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Сообщение: {data}")

Фоновые задачи

from fastapi import BackgroundTasks

def write_notification(email: str, message: str):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, "some notification")
    return {"message": "Notification sent in the background"}

Полный справочник методов и функций FastAPI

Практические примеры использования

API для интернет-магазина

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Интернет-магазин API")

class Product(BaseModel):
    id: Optional[int] = None
    name: str
    price: float
    description: str
    category_id: int

class Category(BaseModel):
    id: Optional[int] = None
    name: str

@app.get("/products/", response_model=List[Product])
async def get_products(category_id: Optional[int] = None, skip: int = 0, limit: int = 100):
    # Логика получения продуктов
    return []

@app.post("/products/", response_model=Product)
async def create_product(product: Product):
    # Логика создания продукта
    return product

@app.get("/categories/", response_model=List[Category])
async def get_categories():
    # Логика получения категорий
    return []

API для блога

from datetime import datetime
from typing import List

class Post(BaseModel):
    id: Optional[int] = None
    title: str
    content: str
    author_id: int
    created_at: datetime = Field(default_factory=datetime.now)
    published: bool = False

class Comment(BaseModel):
    id: Optional[int] = None
    post_id: int
    author_id: int
    content: str
    created_at: datetime = Field(default_factory=datetime.now)

@app.get("/posts/", response_model=List[Post])
async def get_posts(published: bool = True, skip: int = 0, limit: int = 10):
    return []

@app.post("/posts/", response_model=Post)
async def create_post(post: Post):
    return post

@app.get("/posts/{post_id}/comments/", response_model=List[Comment])
async def get_post_comments(post_id: int):
    return []

Часто задаваемые вопросы

В чем основные отличия FastAPI от Flask?

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

Поддерживает ли FastAPI GraphQL?

Да, FastAPI можно интегрировать с GraphQL через библиотеки Strawberry или Graphene.

Можно ли использовать FastAPI с Django ORM?

Да, но это не рекомендуется. Лучше использовать SQLAlchemy, Tortoise ORM или другие асинхронные ORM.

Как обеспечить безопасность FastAPI приложения?

Используйте HTTPS, правильную аутентификацию (JWT, OAuth2), валидацию входных данных, CORS настройки и регулярно обновляйте зависимости.

Поддерживает ли FastAPI миграции базы данных?

FastAPI не включает инструменты для миграций. Используйте Alembic для SQLAlchemy или Aerich для Tortoise ORM.

Можно ли использовать FastAPI для создания полноценных веб-приложений?

FastAPI оптимизирован для создания API. Для полноценных веб-приложений лучше использовать Django или Flask с шаблонами.

FastAPI — это мощный и современный фреймворк, который идеально подходит для создания высокопроизводительных API. Его простота использования, автоматическая документация и отличная производительность делают его отличным выбором для проектов любого масштаба.