Основные концепции FastAPI

FastAPI — это современный, быстрый (высокопроизводительный) Python-фреймворк для создания API. Он основан на стандартных аннотациях типов Python и использует Starlette и Pydantic.

Основные возможности

Высокая производительность: сравнима с NodeJS и Go, один из самых быстрых Python-фреймворков
Быстрая разработка: скорость разработки возросла примерно на 200%-300%
Сокращение ошибок: Сокращение человеческих ошибок примерно на 40%
Интуитивный: Мощная поддержка редакторов, автодополнение повсюду
Простой: Легко использовать и изучать, сокращая время на чтение документации

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

Что такое ASGI?

ASGI (Asynchronous Server Gateway Interface) — это стандартный интерфейс между асинхронными веб-серверами Python и приложениями.

Сравнение WSGI и ASGI:

Пример

# WSGI-приложение (синхронное) — традиционный стиль Flask
def wsgi_app(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [b'Hello World']

# ASGI-приложение (асинхронное) - в стиле FastAPI
async def asgi_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-type', b'text/plain')],
})
await send({
'type': 'http.response.body',
'body': b'Hello World',
})

Основные концепции асинхронного программирования

Синхронное и асинхронное выполнение:

Пример

import asyncio
import time
import httpx

# Синхронный режим - блокирующее выполнение
def sync_fetch_data():
start_time = time.time()
# Имитация трёх сетевых запросов
time.sleep(1) # Первый запрос
time.sleep(1) # Второй запрос
time.sleep(1) # Третий запрос
print(f"Затраченное время для синхронного выполнения: {time.time() - start_time:.2f} seconds") # около 3 seconds

# Асинхронный режим - конкурентное выполнение
async def async_fetch_data():
start_time = time.time()
# Выполнить три запроса одновременно
await asyncio.gather(
asyncio.sleep(1), # Первый запрос
asyncio.sleep(1), # Второй запрос
asyncio.sleep(1), # Третий запрос
)
print(fВремя асинхронного выполнения: {time.time() - start_time:.2f} секунды) # около 1 секунды

# Пример запуска
sync_fetch_data() # Вывод: Время синхронного выполнения: 3.00 секунды
asyncio.run(async_fetch_data()) # Вывод: Время асинхронного выполнения: 1.00 секунды

Асинхронные операции пути в FastAPI:

Пример

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Синхронная операция пути
@app.get("/sync")
def sync_endpoint():
# Синхронные операции блокируют всё приложение
time.sleep(2)
return {"message": "Синхронный ответ"}

# Асинхронная операция пути (рекомендуется)
@app.get("/async")
async def async_endpoint():
# Асинхронные операции не блокируют другие запросы
await asyncio.sleep(2)
return {"message": "Асинхронный ответ"}

# Пример асинхронной операции с базой данных
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Асинхронный запрос к базе данных
user = await database.fetch_one(
"SELECT * FROM users WHERE id = :user_id",
{"user_id": user_id}
)
return user

Когда использовать async?

Сценарии, подходящие для асинхронных операций:

Работа с базой данных
Сетевые запросы (вызовы API)
Операции файлового ввода-вывода
Операции с длительным временем ожидания

Сценарии, не подходящие для асинхронных операций:

Вычисления с высокой нагрузкой на CPU
Простая обработка данных
Операции без ожидания ввода-вывода

Пример

# Хорошее использование async
@app.get("/weather")
async def get_weather():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.weather.com/data")
return response.json()

# Ненужный async (без ожидания I/O)
@app.get("/calculate")
def calculate_sync(): # Просто оставьте синхронным
result = sum(range(1000000))
return {"result": result}

Принципы проектирования REST API

Основы REST

REST (Representational State Transfer) — это стиль проектирования Web API, который делает акцент на:

Ориентированность на ресурсы: URL-адреса представляют ресурсы
Без сохранения состояния: Каждый запрос независим
Единообразный интерфейс: Используйте стандартные методы HTTP
Многоуровневая система: Поддерживает кэширование, балансировку нагрузки и т. д.

Проектирование RESTful URL

Примеры хорошего проектирования URL:

Пример

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

app = FastAPI()

# Коллекции ресурсов и отдельные ресурсы
@app.get("/users") # GET /users - Получить список пользователей
async def get_users():
return users_db

@app.get("/users/{user_id}") # GET /users/123 - Получить конкретного пользователя
async def get_user(user_id: int):
return find_user(user_id)

@app.post("/users") # POST /users - Создать нового пользователя
async def create_user(user: UserCreate):
return create_new_user(user)

@app.put("/users/{user_id}") # PUT /users/123 - Полное обновление пользователя
async def update_user(user_id: int, user: UserUpdate):
return update_existing_user(user_id, user)

@app.patch("/users/{user_id}") # PATCH /users/123 - частичное обновление пользователя
async def patch_user(user_id: int, user: UserPatch):
return patch_existing_user(user_id, user)

@app.delete("/users/{user_id}") # DELETE /users/123 - удаление пользователя
async def delete_user(user_id: int):
return delete_existing_user(user_id)

# Вложенные ресурсы
@app.get("/users/{user_id}/posts") # Получить все статьи пользователя
async def get_user_posts(user_id: int):
return get_posts_by_user(user_id)

@app.post("/users/{user_id}/posts") # Создать новую статью для пользователя
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)

Принципы проектирования URL:

好的设计
GET    /api/v1/users              # 获取用户列表
GET    /api/v1/users/123          # 获取用户 123
POST   /api/v1/users              # 创建用户
GET    /api/v1/users/123/orders   # 获取用户 123 的订单

# 不好的设计
GET    /api/v1/getAllUsers        # 动词不应该在 URL 中
GET    /api/v1/user/123           # 集合名应该用复数
POST   /api/v1/createUser         # URL 中包含动作
GET    /api/v1/users-orders-123   # 关系不清晰

Иерархическая связь ресурсов

Пример

# Проектирование связей между пользователями и статьями
class User(BaseModel):
id: int
name: str
email: str

class Post(BaseModel):
id: int
title: str
content: str
author_id: int

# Основная операция с ресурсом
@app.get("/users")
async def list_users() -> List[User]:
return users

@app.get("/posts")
async def list_posts() -> List[Post]:
return posts

# Операции со связанными ресурсами
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
"""Получить все статьи для конкретного пользователя"""
return [post for post in posts if post.author_id == user_id]

@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
"""Получить информацию об авторе статьи"""
post = find_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Пост не найден")
return find_user(post.author_id)

HTTP-методы и коды состояния

Семантика методов HTTP

Пример

from fastapi import FastAPI, status, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

# GET - безопасный и идемпотентный
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""Получить ресурс без изменения состояния сервера"""
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
return user

# POST - небезопасный и неидемпотентный
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
"""Создать новый ресурс"""
if user_exists(user.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Пользователь уже существует"
)
new_user = create_new_user(user)
return new_user

# PUT - небезопасный, но идемпотентный
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
"""Полностью заменить ресурс"""
if not user_exists(user_id):
# PUT может создать ресурс
return create_user_with_id(user_id, user)
return replace_existing_user(user_id, user)

# PATCH - небезопасен и обычно не идемпотентен
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
"""Частично обновить ресурс"""
existing_user = find_user(user_id)
if not existing_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
return patch_user_fields(existing_user, user)

# DELETE - небезопасен, но идемпотентен
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""Удалить ресурс"""
if not user_exists(user_id):
# Идемпотентность: удаление несуществующего ресурса всё равно возвращает успех
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
delete_existing_user(user_id)
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)

Использование HTTP-кодов состояния

Пример

from fastapi import status

# 2xx успешный ответ
@app.post("/users", status_code=status.HTTP_201_CREATED) # 201 Создано
@app.get("/users") # 200 OK (по умолчанию)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) # 204 Нет содержимого

# 4xx ошибки клиента
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise HTTPException(
status=.HTTP_400_BAD_REQUEST, # 400 Неверный запрос
detail="ID пользователя"
)

user = find_user(user_id)
if not user:
raise HTTPException(
status=.HTTP_404_NOT_FOUND, # 404 Не найдено
detail="Пользователь не найден"
)

return user

# Ошибка бизнес-логики
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
if user_id == current_user_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, # 422 Необрабатываемый объект
detail="Нельзя подписаться на самого себя"
)

if is_already_following(current_user_id, user_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, # 409 Конфликт
detail="Вы уже подписаны на этого пользователя"
)

# Ошибка сервера 5xx (обычно обрабатывается обработчиком исключений)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # 500 Внутренняя ошибка сервера
content={"detail": "Внутренняя ошибка сервера"}
)

Краткий справочник по распространённым кодам состояния:

# 成功状态码
200 OK                    # 请求成功
201 Created              # 资源创建成功
204 No Content           # 成功但无内容返回
206 Partial Content      # 部分内容（分页、断点续传）

# 重定向
301 Moved Permanently    # 永久重定向
302 Found               # 临时重定向
304 Not Modified        # 资源未修改（缓存）

# 客户端错误
400 Bad Request         # 请求格式错误
401 Unauthorized        # 未认证
403 Forbidden          # 已认证但无权限
404 Not Found          # 资源不存在
405 Method Not Allowed  # HTTP 方法不允许
409 Conflict           # 资源冲突
422 Unprocessable Entity # 数据验证失败
429 Too Many Requests   # 请求过于频繁

# 服务器错误
500 Internal Server Error # 服务器内部错误
502 Bad Gateway          # 网关错误
503 Service Unavailable  # 服务不可用

Формат данных JSON

Основы JSON и лучшие практики

Пример

from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json

# Спецификация формата данных JSON
class UserResponse(BaseModel):
id: int
username: str
email: str
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None

# Конфигурация сериализации JSON
class Config:
# Разрешить псевдонимы полей
allow_population_by_field_name = True
# Кодировщик JSON
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}

# Стандартизировать формат ответа
class APIResponse(BaseModel):
"""Стандартный формат ответа API"""
success: bool = True
message: str = "Операция успешно выполнена"
data: Optional[dict] = None
errors: Optional[List[str]] = None
timestamp: datetime = Field(default_factory=datetime.now)

# Формат ответа пагинации
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
size: int
pages: int

@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
"""Возвращает постраничный список пользователей"""
users = get_users_paginated(page, size)
total = count_users()

return PaginatedResponse(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)

JSON Schema и валидация данных

Пример

from pydantic import BaseModel, validator, Field
from typing import Optional, List
from enum import Enum

class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"

class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="Имя пользователя")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="Адрес электронной почты")
password: str = Field(..., min_length=8, description="Пароль")
full_name: Optional[str] = Field(None, max_length=100, description="Полное имя")
role: UserRole = Field(UserRole.USER, description="Роль пользователя")
age: Optional[int] = Field(None, ge=0, le=150, description="Возраст")

@validator('username')
def username_must_be_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('Имя пользователя может содержать только буквы и цифры')
return v

@validator('password')
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Пароль должен содержать хотя бы одну заглавную букву')
if not any(c.islower() for c in v):
raise ValueError('Пароль должен содержать хотя бы одну строчную букву')
if not any(c.isdigit() for c in v):
raise ValueError('Пароль должен содержать хотя бы одну цифру')
return v

class Config:
# Пример генерации JSON Schema
schema_extra = {
"example": {
"username": "johndoe",
"email": "[email protected]",
"password": "SecurePass123",
"full_name": "John Doe",
"role": "user",
"age": 30
}
}

# Используйте пользовательские валидаторы
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Создать нового пользователя, включая проверку данных"""
# Pydantic автоматически проверяет входные данные
return await create_new_user(user)

Формат ответа с ошибкой

Пример

class ErrorDetail(BaseModel):
field: str
message: str
code: str

class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[List[ErrorDetail]] = None
timestamp: datetime = Field(default_factory=datetime.now)

# Пользовательская обработка исключений
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append(ErrorDetail(
field='.'.join(str(x) for x in error['loc']),
message=error['msg'],
code=error['type']
))

return JSONResponse(
status_code=422,
content=ErrorResponse(
error="Ошибка валидации",
message="Ошибка валидации данных запроса",
details=errors
).dict()
)

@app.exception_handler⟦⟧StarletteHTTPException(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
message=exc.detail
).dict()
)

Концепции, специфичные для FastAPI

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

Пример

from fastapi import FastAPI, Query, Path, Body
from fastapi.openapi.utils import get_openapi

app = FastAPI(
title="Мой API",
description=Это пример API, демонстрирующий возможности FastAPI,
version="1.0.0",
terms_of_service="http://example.com/terms/",
contact={
"name": "Разработчик",
"url": "http://example.com/contact/",
"email": "[email protected]",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
)

@app.get(
"/users/{user_id}",
summary=Получить информацию о пользователе,
description="Получить подробную информацию о пользователе по ID пользователя",
response_description=Объект информации о пользователе,
tags=["Управление пользователями"]
)
async def get_user(
user_id: int = Path(..., title="ID пользователя", description="ID пользователя для получения", ge=1),
include_posts: bool = Query(False, title="Включить статьи", description=Включать ли список статей пользователя)
):
"""
Получить информацию о пользователе:

- **user_id**: Уникальный идентификатор пользователя.
**include_posts**: Включать ли список публикаций пользователя в ответ

Вернуть основную информацию о пользователе; если include_posts имеет значение True, в ответ также будет включён список статей.
"""
user = find_user(user_id)
if include_posts:
user.posts = get_user_posts(user_id)
return user

# Пользовательская схема OpenAPI
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema

openapi_schema = get_openapi(
title="Пользовательская документация API",
version="2.5.0",
description="Это пользовательская схема OpenAPI",
routes=app.routes,
)

# Добавить пользовательскую информацию
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}

app.openapi_schema = openapi_schema
return app.openapi_schema

app.openapi = custom_openapi

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

Пример

from fastapi import Depends

# Простая зависимость
def get_current_user_id() -> int:
# Извлечь ID пользователя из токена
return 123

def get_database_session():
# Создать сессию базы данных
db = DatabaseSession()
try:
yield db
finally:
db.close()

# Использование зависимостей
@app.get("/profile")
async def get_profile(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return get_user_profile(db, user_id)

# цепочка зависимостей
def get_current_user(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return db.query(User).filter(User.id == user_id).first()

@app.get("/dashboard")
async def get_dashboard(
current_user: User = Depends(get_current_user) # Зависит от других зависимостей
):
return generate_dashboard(current_user)

Важность аннотаций типов

Пример

from typing import List, Optional, Union, Dict, Any
from pydantic import BaseModel

# FastAPI полностью полагается на аннотации типов:
# 1. Валидация данных
# 2. Генерация документации
# 3. Поддержка редактора

@app.get("/items/{item_id}")
async def get_item(
item_id: int, # Параметр пути, автоматически преобразуется в int
q: Optional[str] = None, # Необязательные параметры запроса
limit: int = 10 # Параметры запроса со значениями по умолчанию
) -> Dict[str, Any]: # Аннотация типа возвращаемого значения
"""Аннотации типов сообщают FastAPI, как обрабатывать параметры и ответы"""
return {"item_id": item_id, "q": q, "limit": limit}

# Сложная аннотация типа
@app.post("/items/")
async def create_items(
items: List[ItemCreate] # Получает список объектов ItemCreate
) -> List[ItemResponse]: # Возвращает список объектов ItemResponse
return [create_item(item) for item in items]

# Тип Union (несколько возможных типов)
@app.get("/search")
async def search(
q: str,
result_type: str = "json"
) -> Union[List[Dict], str]: # Может возвращать словарь, список или строку
if result_type == "json":
return [{"title": "Result 1"}, {"title": "Результат 2"}]
else:
return "Результат 1, Результат 2"

Практические примеры основных концепций

Пример

from fastapi import FastAPI, HTTPException, status, Depends, Query
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio

# Комплексный пример: API блога
app = FastAPI(
title=API блога,
description="Система блога, демонстрирующая основные концепции FastAPI",
version="1.0.0"
)

# Модель данных
class PostBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
published: bool = Field(True)

class PostCreate(PostBase):
pass

class PostResponse(PostBase):
id: int
author_id: int
created_at: datetime
updated_at: Optional[datetime] = None

class Config:
orm_mode = True

# Мок-база данных
posts_db = []
next_id = 1

# Зависимость: получить текущего пользователя (упрощённая версия)
async def get_current_user() -> int:
# В реальных приложениях он будет извлечён из JWT-токена
return 1

# Асинхронная операция пути
@app.get("/posts", response_model=List[PostResponse], tags=["Статья"])
async def list_posts(
skip: int = Query(0, ge=0, description=Количество пропущенных статей),
limit: int = Query(10, ge=1, le=100, description=Количество возвращаемых статей),
published_only: bool = Query(True, description="Возвращать только опубликованные статьи")
):
"""
Получить список статей

Поддержка пагинации и фильтрации
"""
# Имитация асинхронного запроса к базе данных
await asyncio.sleep(0.1)

filtered_posts = posts_db
if published_only:
filtered_posts = [p for p in posts_db if p["published"]]

return filtered_posts[skip:skip + limit]

@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["Article"])
async def create_post(
post: PostCreate,
current_user_id: int = Depends(get_current_user)
):
"""
Создать новую статью

Требуется аутентификация пользователя
"""
global next_id

# Имитация асинхронных операций с базой данных
await asyncio.sleep(0.1)

new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"published": post.published,
"author_id": current_user_id,
"created_at": datetime.now(),
"updated_at": None
}

posts_db.append(new_post)
next_id += 1

return new_post

@app.get("/posts/{post_id}", response_model=PostResponse, tags=["Статья"])
async def get_post(post_id: int):
"""
Получить конкретную статью

Возвращает информацию о статье по ID статьи
"""
# Имитация асинхронного запроса к базе данных
await asyncio.sleep(0.1)

post = next((p for p in posts_db if p["id"] == post_id), None)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Пост с id {post_id} не найден"
)

return post

# Эндпоинт проверки работоспособности
@app.get("/health", tags=[System])
async def health_check():
"""Проверка работоспособности системы"""
return {
"status": "healthy",
"timestamp": datetime.now(),
"posts_count": len(posts_db)
}

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

Другие расширения

Практический проект

Основные концепции FastAPI

Основные возможности

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

Что такое ASGI?

Пример

Основные концепции асинхронного программирования

Пример

Пример

Когда использовать async?

Пример

Принципы проектирования REST API

Основы REST

Проектирование RESTful URL

Пример

Иерархическая связь ресурсов

Пример

HTTP-методы и коды состояния

Семантика методов HTTP

Пример

Использование HTTP-кодов состояния

Пример

Формат данных JSON

Основы JSON и лучшие практики

Пример

JSON Schema и валидация данных

Пример

Формат ответа с ошибкой

Пример

Концепции, специфичные для FastAPI

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

Пример

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

Пример

Важность аннотаций типов

Пример

Практические примеры основных концепций

Пример