Ускоряем разработку на FastAPI: Мощный инструментарий для создания и тестирования API
В мире современной веб-разработки FastAPI зарекомендовал себя как мощный и быстрый фреймворк для создания API. Однако, при работе над крупными проектами разработчики часто сталкиваются с необходимостью оптимизировать рутинные процессы, улучшить структуру кода и упростить тестирование. В этой статье мы рассмотрим набор инструментов, который поможет решить эти задачи и существенно ускорить разработку на FastAPI.
Ходя по множеству собеседований, я заметил что многие компании, активно использующие FastAPI, разработали собственные библиотеки, но аналогичных инструментов с открытым доступом и свободной лицензией практически нет. Я, как опытный backend-разработчик на Python и Django, решил адаптировать и интегрировать наиболее полезные и востребованные решения для разработки REST API в FastAPI, основываясь на своем опыте работы с Django.
Краткосрочная цель: собрать обратную связь от сообщества о данной идее.
Долгосрочная цель: улучшить инструмент в open source, побуждая крупные компании отказаться от разработки и поддержки собственного проприетарного кода. Вместо этого мы стремимся создать экосистему, где компании не только используют общий инструментарий, но и активно участвуют в его совершенствовании, внося свой вклад в open-source проект.
Для кого это будет полезно?
Backend-разработчикам на Python, использующим или планирующим использовать FastAPI
Командам, работающим над средними и крупными проектами на FastAPI
Разработчикам, которые хотят улучшить структуру своих FastAPI-проектов и ускорить процесс разработки
Тем, кто ищет эффективные инструменты для тестирования FastAPI-приложений
Зачем нужен этот инструментарий?
FastAPI Accelerator — это open-source инструментарий, созданный на основе лучших практик разработки REST API.
Основная цель представленного инструментария — ускорить и упростить разработку проектов на FastAPI.
Это достигается путем:
Подробной и хорошей документацией.
Предоставления переиспользуемого кода для типовых задач.
Внедрения универсального менеджера для работы с РСУБД.
Реализации ViewSet для быстрого создания представлений с базовой бизнес-логикой.
Интеграции аутентификации по JWT.
Добавления удобной админ-панели.
Упрощения написания и выполнения интеграционных тестов для API.
Оптимизации работы с Alembic для управления миграциями в production и test окружениях.
Все эти компоненты взаимосвязаны и дополняют друг друга, автоматизируя рутинные задачи.
Структура инструмента
Давайте рассмотрим основные компоненты нашего инструментария:
fastapi_accelerator/
├── db/ # Логика взаимодействия с РСУБД
├── pattern/ # Шаблоны для проектов
├── testutils/ # Утилиты для тестирования FastAPI
├── cache.py # Реализация кеширования
├── auth_jwt.py # Аутентификация по JWT
├── exception.py # Обработка исключений
├── middleware.py # Middleware компоненты
├── paginator.py # Реализация пагинации
├── timezone.py # Работа с временными зонами
├── viewset.py # Реализация ViewSet
└── utils.py # Общие утилитыСтруктуры проекта FastAPI
Правильная организация проекта — ключ к его масштабируемости и удобству поддержки. Вот пример рекомендуемой структуры проекта FastAPI с использованием FastAPI Accelerator:
Проект/
│
├── app/
│ ├── __init__.py
│ ├── utils.py # Пере используемый функционал для проекта
│ ├── core/ # Содержит основные модули, такие как конфигурация, безопасность и общие зависимости.
│ │ ├── __init__.py
│ │ ├── settings_local.py # Локальные значения для настроек, не должны быть в git, создавать непосредственно на сервере
│ │ ├── config.py # Настройки проекта которые не зависят от внешних настроек
│ │ ├── security.py # Логика безопасности проекта
│ │ ├── db.py # Настройки и сессии базы данных.
│ │ ├── cache.py # Настройки кеширования
│ │ └── dependencies.py
│ │
│ ├── api/ # Содержит все API эндпоинты, разделенные по версиям.
│ │ ├── __init__.py
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── router.py # Содержит обработчики запросов для указанной версии api
│ │ │
│ │ ├── static/ # Содержит файлы статики, если они нужны
│ │ │ ├── js
│ │ │ ├── css
│ │ │ ├── img
│ │ │ └── html
│ │ │
│ │ ├── logic/ # Содержит бизнес логику
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── items.py
│ │ │
│ │ ├── schemas/ # Pydantic модели для валидации запросов и ответов.
│ │ │ ├── __init__.py
│ │ │ ├── user.py
│ │ │ └── item.py
│ │ │
│ │ ├── crud/ # Функции для работы с базой данных (Create, Read, Update, Delete).
│ │ │ ├── __init__.py
│ │ │ ├── user.py
│ │ │ └── item.py
│ │ │
│ │ └── tests/ # Директория для тестов.
│ │ ├── __init__.py
│ │ ├── test_users.py
│ │ └── test_items.py
│ │
│ ├── models/ # Определения моделей базы данных (например, SQLAlchemy модели).
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ │
│ └── fixture/ # Хранит фикстуры для тестирования этого проекта
│ ├── __init__.py
│ ├── items_v1.py # Тестовые записи для БД
│ └── utils.py # Переиспользуемые фикстуры для тестов
│
├── fastapi_accelerator/ # Submodule для переиспользовать
│
├── alembic/ # Директория для миграций базы данных.
│ ├── versions/ # Папка с миграциями
│ │ ├── __init__.py
│ │ └── 0001_init.py # Файл с миграцией
│ └── env.py # Настройки для alembic
│
├─ conf/ # Файлы конфигурации для prod
│ ├── settings_local.example.py # Пример для создания settings_local.py
│ └── Dockerfile # Файл для prod
│
├── pytest.ini # Конфигурация для pytest
├── conftest.py # Настройки выполнения тестов
│
├── .gitignore # Какие игнорировать файлы и папки в git
├── .gitlab-ci.yml # Настройки CI pipeline
│
├── pyproject.toml # Настройки Poetry
│
├── Makefile # Переиспользуемые bash команды
│
├── README.md # Описание проекта
├── CHANGELOG.md # Изменения в проекте
├── version.toml # Версия проекта
│
├── alembic.ini # Конфигурации для alembic
│
├── DockerfileDev # Файл для создания dev контейнера с APP
├── docker-compose.yml # Используется для сборки dev окружения
│
├── admin_panel.py # Админ панель
│
└── main.py # Точка входа в приложение, где создается экземпляр FastAPI.Подключение в FastAPI
Файл main.py:
from fastapi import FastAPI
from fastapi_accelerator.pattern.pattern_fastapi import base_pattern
from app.core.config import BASE_DIR_PROJECT, DEBUG, SECRET_KEY
from fastapi_accelerator.timezone import moscow_tz
from app.core.db import DatabaseManager
from app.core.security import AuthJWT
import app.api.v1.router as RouterV1
app = FastAPI()
# Паттерн для проекта
base_pattern(
app,
routers=(RouterV1.router,),
timezone=moscow_tz,
cache_status=True,
debug=DEBUG,
base_dir=BASE_DIR_PROJECT,
database_manager=DatabaseManager,
secret_key=SECRET_KEY,
)
# Подключить аутентификацию по JWT
AuthJWT.mount_auth(app)Основные компоненты
Base Pattern
Функция base_pattern добавляет множество полезных функций в app, включая:
Заполнение
stateи другую информацию уapp.Разрешение
CORS.Подключение роутеров с поддержкой
ViewSet.Добавление метода
healthcheck.Middlewareдля отладки времени выполнения API-запросов.Подробный вывод для
HTTPисключений.
DatabaseManager
DatabaseManager — это универсальный инструмент для работы с РСУБД, предоставляющий как синхронные, так и асинхронные (название начинается на a) методы. DatabaseManager использует патер одиночка, поэтому может быть легко подменен в тестах.
Пример использования:
from app.core.config import DATABASE_URL, DEBUG, DEV_STATUS
from fastapi_accelerator.dbsession import MainDatabaseManager
DatabaseManager = MainDatabaseManager(DATABASE_URL, echo=DEBUG, DEV_STATUS=DEV_STATUS)Общие характеристики
DEV_STATUS— Индикатор режима разработки. ПриDEV_STATUS=Falseблокирует выполнение критических операций (create_all,drop_all,clear_all). Это мера безопасности для производственной среды.
Синхронные компоненты
database_url— Адрес для подключения к синхронной базе данных.engine— Механизм синхронного взаимодействия с БД.session— Генератор синхронных сессий.Base— Базовый класс для моделей данных.Функциональность:
get_session— Инжектор сессии БД.get_session_transaction— Инжектор сессии БД с поддержкой транзакций.create_all— Инициализация всех таблиц в БД.drop_all— Удаление всей структуры БД.clear_all— Очистка содержимого таблиц. Параметрexclude_tables_nameпозволяет исключить определенные таблицы из процесса очистки.
Асинхронные компоненты
adatabase_url— Адрес для подключения к асинхронной БД.aengine— Асинхронный механизм работы с БД, включая пул соединений.asession— Генератор асинхронных сессий.Функциональность:
aget_session— Асинхронный инжектор сессии БД.aget_session_transaction— Асинхронный инжектор сессии БД с поддержкой транзакций.
OrmAsync
Этот класс оптимизирует асинхронное взаимодействие с БД:
get— Извлечение объекта по заданным критериям.get_list— Получение набора объектов по запросу. (С возможностью глубокой выборки)update— Модификация объектов согласно запросу.delete— Удаление объектов по заданным параметрам.get_item— Извлечение объекта по первичному ключу. (С возможностью глубокой выборки)create_item— Создание нового объекта. (С возможностью каскадного создания)update_item— Обновление объекта по первичному ключу. (С возможностью каскадного обновления)delete_item— Удаление объекта по первичному ключу. (С возможностью каскадного удаления)
Глубокая выборка/каскадные операции — это возможность работы со связанными данными.
Активируется параметромdeep=TrueПримеры:
get_list, get_item — Возвращают объекты со всеми связанными данными, готовые для использования в Pydantic
create_item — Создает записи в связанных таблицах
update_item — Обновляет данные в связанных таблицах
delete_item — Удаляет записи из связанных таблиц
ViewSet
ViewSet позволяет быстро создавать CRUD-операции для моделей. Вот пример использования:
from fastapi_accelerator.viewset import AppOrm, FullViewSet
from fastapi import APIRouter, Depends, Query
from app.api.v1.schemas.timemeasurement import TaskExecution
from app.models.timemeasurement import TaskExecution as TaskExecutionDb
router = APIRouter(prefix="/api/v1")
class FileViewSet(FullViewSet):
"""
Представление для работы с файлами
"""
# Модель БД
db_model = TaskExecutionDb
# Модель Схемы
pydantic_model = TaskExecution
'''
# Кеширование
cache_class = redis_client
cache_ttl = timedelta(minutes=10)
# Пагинация
paginator_class = DefaultPaginator
# Включить поддержку вложенных схем pydantic
# это значит что будет происходить рекурсивное
# создание, обновление, удаление связанных записей
deep_schema = True
# Включить защиту через JWT
dependencies = [Depends(jwt_auth)]
# Вы можете также переопределять методы:
async def db_update(
self, item_id: str | int | UUID, item: type[BaseModel], aorm: OrmAsync
) -> object:
"""Переопределение метода db_update"""
return await super().db_update(item_id, item, aorm)
def list(self):
"""Переопределение метода list"""
@self.router.get(f"{self.prefix}", tags=self.tags)
async def get_list_items(
skip: int = Query(0),
limit: int = Query(100),
aorm: OrmAsync = Depends(AppOrm.aget_orm),
) -> List[self.pydantic_model]:
return await aorm.get_list(
self.db_model,
select(self.db_model).offset(skip).limit(limit),
deep=self.deep_schema,
)
return get_list_items
'''
router.views = [
FileViewSet().as_view(router, prefix="/file"),
]Аутентификация по JWT
Для защиты API-эндпоинтов мы используем JWT-аутентификацию:
from fastapi_accelerator.auth_jwt import BaseAuthJWT
class AuthJWT(BaseAuthJWT):
def check_auth(username: str, password: str) -> bool:
"""Проверка введенного логина и пароля."""
return username == "admin" and password == "admin"
AuthJWT.mount_auth(app)Пример защиты API метода:
from fastapi_accelerator.auth_jwt import jwt_auth
@app.get("/check_protected", summary="Проверить аутентификацию по JWT")
async def protected_route(jwt: dict = Depends(jwt_auth)):
return {"message": "This is a protected route", "user": jwt}Админ-панель
Для удобного управления данными мы интегрировали Flask-Admin:
from flask import Flask
from app.core.config import ADMIN_PASSWORD, ADMIN_USERNAME, SECRET_KEY
from app.db.base import DatabaseManager
from app.models import File, User
from fastapi_accelerator.pattern_flask_admin import base_pattern
app = Flask(__name__)
admin = base_pattern(
app,
SECRET_KEY,
ADMIN_PASSWORD,
ADMIN_USERNAME,
# > Модели которые нужны в админ панели
models=[User, File],
database_manager=DatabaseManager,
)
if __name__ == "__main__":
app.run(
host="0.0.0.0",
port=8001,
debug=True,
)Тестирование
Одной из ключевых особенностей нашего инструментария является мощная система для написания и выполнения тестов. Она включает в себя:
Фикстуры для работы с тестовой базой данных и клиентом API.
Декораторы для аутентификации и применения фикстур.
Контекстный менеджер для отслеживания SQL-запросов.
Утилиты для проверки JSON-ответов.
Тестирование через классы.
Пример функции теста:
from typing import Callable, NamedTuple
from fastapi.testclient import TestClient
from app.fixture.items_v1 import export_fixture_file
from fastapi_accelerator.db.dbsession import MainDatabaseManager
from fastapi_accelerator.testutils import apply_fixture_db, client_auth_jwt, track_queries, check_response_json
# Аутентифицировать тестового клиента
@client_auth_jwt(username="test")
# Создать тестовые данных из функции с фикстурами
@apply_fixture_db(export_fixture_file)
def test_имя(
client: TestClient, # Тестовый клиент для API запросов
url_path_for: Callable, # Функция для получения url по имени функции обработчика
db_manager: MainDatabaseManager, # Менеджер тестовой БД
fixtures: NamedTuple, # Хранит созданные данные из фикстур
):
# Проверка количество выполняемых SQL команд
with track_queries(db_manager, expected_count=3):
# Запрос в API
response = client.get(url_path_for("ИмяФункции"))
# Проверка JSON API ответа
check_response_json(
response,
200,
{
"id": fixtures.Имя.id,
},
)Пример тест класса:
from typing import Callable, NamedTuple
from fastapi.testclient import TestClient
from app.fixture.items_v1 import export_fixture_file
from fastapi_accelerator.db.dbsession import MainDatabaseManager
from fastapi_accelerator.testutils import apply_fixture_db
from fastapi_accelerator.testutils.fixture_auth import client_auth_jwt
from fastapi_accelerator.testutils.fixture_db.trace_sql import track_queries
from fastapi_accelerator.testutils.utils import BaseAuthJwtPytest, check_response_json
BASE_URL_V1 = "/api/v1/"
class TestИмя(BaseAuthJwtPytest):
# Создать тестовые данных из функции с фикстурами
@apply_fixture_db(export_fixture_file)
def setUp(self, fixtures: NamedTuple):
self.url = BASE_URL_V1 + "taskexecution"
self.fixtures = fixtures # Хранит созданные данные из фикстур
def test_имя(self, client: TestClient, db_manager: MainDatabaseManager):
# Проверка количество выполняемых SQL команд
with track_queries(db_manager, expected_count=3):
# Запрос в API
response = client.get(self.url)
# Проверка JSON API ответа
check_response_json(
response,
200,
{
"id": self.fixtures.Имя.id,
},
)Сравнение с существующими решениями
Хотя существует несколько проектов, предлагающих инструменты для разработки и тестирования FastAPI приложений, наше решение выделяется своей комплексностью и специализацией:
FastAPI-Utils: Предоставляет утилиты для разработки, но менее фокусирован на тестировании.FastAPI-SQLAlchemy: Интегрирует FastAPI с SQLAlchemy, включая некоторые утилиты для тестирования.FastAPI-Toolkit: Предлагает набор инструментов, но менее специализирован на задачах тестирования.freddie— В архиве на GitHub, только viewsetfastapi_viewsets— Только viewsetFastAPIwee— Менее специализирован на задачах тестирования.
Наше решение отличается тем, что:
Более специфично для задач тестирования FastAPI приложений.
Предоставляет более широкий набор инструментов для различных аспектов тестирования.
Включает уникальные функции, такие как декоратор
@apply_fixture_dbи контекстный менеджерtrack_queries.Предлагает комплексный подход, охватывающий различные аспекты разработки и тестирования FastAPI приложений.
Заключение
Представленный инструментарий значительно упрощает и ускоряет разработку на FastAPI. Он предоставляет готовые решения для типовых задач, улучшает структуру проекта и облегчает тестирование. Использование этих инструментов позволит разработчикам сосредоточиться на бизнес-логике приложения, а не на технических деталях реализации.
Несмотря на наличие других инструментов в экосистеме FastAPI, наше решение выделяется своей полнотой и специализацией именно на задачах тестирования. Это делает его ценным дополнением к существующим ресурсам для разработчиков FastAPI.
Мы продолжаем развивать этот инструментарий и будем рады обратной связи от сообщества. Если у вас есть идеи по улучшению или вы нашли ошибку, пожалуйста, создайте issue в нашем репозитории на GitHub.
Планы развития
Обеспечить совместимость с последними (прошлыми) версиями FastAPI и связанных библиотек
Значительно увеличить покрытие кода тестами
Оптимизировать существующий код для повышения производительности
Провести нагрузочное тестирование и оптимизировать критические участки кода
Провести рефакторинг с учетом лучших практик и паттернов проектирования
Разработать и добавить аналогичный вариант для работы с
WebSocketУсовершенствовать механизм выполнения фоновых задач (
background tasks)Улучшить реализацию периодических задач, подобных тем, что представлены в
fastapi-utilsСоздать несколько детальных примеров проектов, демонстрирующих различные сценарии использования
Расширить документацию, добавив больше практических руководств и рекомендаций по применению
Исследовать возможности интеграции с популярными инструментами экосистемы FastAPI
