Около полутора лет назад я представил вниманию читателей статью, где описал свое видение структуры FastAPI-приложения, поделился опытом выстраивания архитектуры системы, а также показал свой репозиторий-шаблон для старта проектов. Я стал старше, опытнее, набил шишки в ходе разработки сразу двух крупных проектов, рефакторинга как своего, так и чужого кода и выступил лидом для команды джунов, а это значит, что пора поделиться своим опытом с миром.
В новой статье я хочу поговорить о структуре и архитектуре приложения, рассказать, с какими трудностями я столкнулся в ходе разработки production-системы, а также показать практики, которые каждый день облегчают мне разработку. Приятного чтения!
Инфраструктура
Удобная и логичная инфраструктура значительно ускоряет процесс разработки, деплоя и поддержки продукта. Разворачивать проект должно быть просто и удобно, особенно локально. Если для запуска разработчику каждый раз приходится идти в огромный README.md и собирать команду из нескольких параметров, инфраструктуру уже стоит немного упростить.
Разные контуры — разная инфраструктура
Помимо локально развернутых контейнеров бэкенда, базы и т.п. часто есть и другие контуры приложения — dev, stage, production, которые необходимо запускать и поддерживать. Прод может ходить в облачное S3-хранилище для сохранения или получения аватарок пользователей, в то время как dev себе такого позволить не может, ведь хранение файлов в облаке — это деньги. Для локальной разработки и dev-контура вполне логично и нормально использовать MinIO в качестве файлового хранилища, а для прода этот контейнер уже не нужен. Возникает неочевидная на первый взгляд проблема — разделение инфраструктуры для разных контуров.
Сначала я пытался решать эту проблему разными .env файлами и запуском Docker Compose с необходимыми параметрами, но это быстро превратилось в путаницу между env-шаблонами, compose-файлами и переменными окружения. Решение оказалось простым и удобным — разные compose-файлы для разных контуров. В итоге у меня получилось:
-
local.yml— база,MinIO,Stripe -
development.yml— база, бэкенд,MinIO,Stripe,Nginx -
production.yml— база, бэкенд,Nginx
Для всех конфигов общая только база данных, для разных контейнеров можно прописывать свои переменные окружения прямо в конфигурации, а общие вынести в .env файл. Таким образом, мы практически унифицируем env-файл и снимаем с себя необходимость помнить profiles или имена контейнеров для запуска приложения в различных сценариях. Теперь нам достаточно знать только имя compose-файла:
# Локальный контурdocker compose -f local.yml up --build -d# Dev-контурdocker compose -f development.yml up --build -d# Prod-контурdocker compose -f production.yml up --build -d
Делайте скрипты
Один старый мудрый программист часто мне говорит: «Увидел умную мысль — скопируй». Этот совет в сумме с моим опытом привел меня к тому, что я начал писать скрипты. Сделать бэкап БД — скрипт, обновить SSL-сертификаты — скрипт. Скрипт на какое-то действие удобнее, чем вспоминать, какие команды там надо писать. Плюс ко всему, хороший скрипт — это хорошая заметка.
Кешируйте слои при билде образа
Docker кеширует слои при билде образа, поэтому стоит один раз потратить время на эффективный Dockerfile. Правильный порядок слоев может заметно сократить время последующих сборок, что особенно ощутимо, например, для Next.js.
Остальное
Можно также поговорить про CI/CD и тесты. Этим вещам однозначно есть место в разговоре об инфраструктуре, однако о них и так все знают. Мой главный тейк по этому вопросу заключается в том, что нужно проводить четкую границу между тем, что просто было бы здорово использовать эти вещи и тем, что без них становится трудно. Развертывание и поддержка dev-контура — это однозначно тот момент, когда проще потратить время на написание пайплайна, чем рано или поздно поймать себя на мысли:
Что-то мне надоело каждое обновление в dev-ветке разворачивать еще и на серваке
С тестами такая же история. Для маленького приложения покрытие кода тестами может долго оставаться незаметным, однако рост приложения неизбежно ведет к появлению связности компонентов системы, а значит изменения в одном месте могут уронить работу кода в другом. Писать тесты лениво и долго, поэтому предлагаю скинуть это на ИИ-агента. Не бездумно, конечно, вы же не хотите потом ДЕБАЖИТЬ ТЕСТЫ.
FastAPI
Я использую Domain-Driven Design для проектирования ПО вокруг бизнес-правил и чистую архитектуру для организации слоев приложения. На эту тему написано уже достаточно большое количество материалов и статей, поэтому рассказывать что это и как это я не буду. Я хочу поделиться своим опытом следования этим подходам в разработке, затронуть тему границ слоев и поделиться своими фишками, которые мне облегчают процесс разработки.
Работа с БД, транзакциями и бизнес-логикой
Велосипед я не изобретал — использую репозитории, как обертку над БД, свой менеджер транзакций, найденный на просторах интернета. Бизнес-логику выполняю в сервисах, а несколько бизнес-действий выполняю уже в юз кейсах.
Репозиторий позволяет мне выполнять простейшие действия с базой, причем эти действия я закрепляю контрактом, чтобы при необходимости можно было легко заменить одну реализацию репозитория на другую. Базовый репозиторий:
class BaseRepository(ABC, Generic[T]): @abstractmethod async def get_first_by_kwargs(self, **kwargs) -> T | None: pass @abstractmethod async def get_all_by_kwargs(self, **kwargs) -> Sequence[T]: pass @abstractmethod async def list(self) -> tuple[Sequence[T], int]: pass @abstractmethod async def create(self, **kwargs) -> T: pass @abstractmethod async def update(self, object_id: int, **kwargs) -> T: pass @abstractmethod async def remove(self, object_id: int) -> Any: pass
Я допустил небольшое расхождение имени метода list и его возвращаемого значения. Я реализую метод list специально для работы с сортировкой, фильтрацией и пагинацией, а название осталось историческое.
Этот контракт меняется крайне редко, поэтому является хорошей основой для разных реализаций репозитория. Например, одна реализация может работать с объектами SQLAlchemy, другая — с Qdrant PointStruct. К примеру, имплементация метода get_first_by_kwargs для SQLAlchemyRepository:
class SQLAlchemyRepository(BaseRepository[T], Generic[T]): model: type[T] def __init__(self, session: AsyncSession): self.session = session async def get_first_by_kwargs(self, stmt=None, **kwargs) -> T | None: if stmt is None: stmt = select(self.model) stmt = stmt.filter_by(**kwargs) return (await self.session.execute(stmt)).scalars().first()
Новые методы добавляются скорее в разных реализациях базового репозитория, чем как-то меняют контракт. SQLAlchemyRepository позволяет не дублировать CRUD-логику для моделей за счет банального наследования:
class UserRepository(SQLAlchemyRepository): model = User
Вы можете сказать, что для каждой сущности создавать свой репозиторий это неудобно, и будете в каком-то смысле правы, однако такой подход отлично раскрывается с менеджером транзакций, который аккумулирует в себе репозитории:
class BaseTransactionManager(ABC): @abstractmethod async def __aenter__(self): ... @abstractmethod async def __aexit__(self, exc_type, exc_val, exc_tb): ... @abstractmethod async def rollback(self): ... @abstractmethod async def commit(self): ...class SQLAlchemyTransactionManagerBase(BaseTransactionManager): def __init__(self, session: AsyncSession): self._session = session async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): if exc_type: await self.rollback() else: await self.commit() await self._session.close() async def rollback(self): await self._session.rollback() async def commit(self): await self._session.commit()
Я пробовал большое количество решений для того, чтобы сделать работу с транзакциями удобной:
-
Делал свою реализацию менеджера транзакций на каждый домен, собирая только те репозитории, с которыми работал в текущий момент;
-
Пытался отказаться от менеджера транзакций вообще, но это лишало меня возможности работать с единой транзакцией для разных сущностей;
-
Пытался работать именно с репозиториями и прокидывал сессии сразу в них, но это превращалось в костыли в сервисах.
Решение оказалось достаточно простым — собрать репозитории внутри менеджера:
class TransactionManager(SQLAlchemyTransactionManagerBase): @property def user_repository(self): return UserRepository(self._session) @property def job_repository(self): return JobRepository(self._session)def get_transaction_manager( session: Annotated[AsyncSession, Depends(session_getter)],) -> BaseTransactionManager: return TransactionManager(session)TransactionManagerDep = Annotated[TransactionManager, Depends(get_transaction_manager)]
Интуитивное решение, которое работает неинтуитивно. Сессия живет внутри инстанса менеджера, а значит все репозитории, работающие через менеджер, вызывают свои методы в рамках одной сессии, что отлично ложится в отдельные сервисы/эндпоинты/юз кейсы. Есть несколько стандартных сценариев использования, выглядят они примерно так:
# напрямую выполнить вызов из репозиторияasync with transaction_manager_instance: users, count = await transaction_manager_instance.user_repository.list( limit, offset, order_by, filters )# оркестрация бизнес-действий в сценарияхclass Service: async def first_service_method(): user = await transaction_manager_instance.user_repository.get_first_by_kwargs( user_email="coolguy2012@example.com", ) user_job = await transaction_manager_instance.job_repository.create( job="Python developer", user_id=user.id, ) async def second_service_method(): user_job = await transaction_manager_instance.job_repository.remove( object_id=123, )async def use_case(): async with transaction_manager_instance: await service.first_service_method() await service.second_service_method()
Такой подход имеет сразу пару существенных плюсов:
-
своя надстройка над управлением транзакцией — сервисный слой становится проще, приходится писать меньше boilerplate-кода;
-
нам не нужно думать, откуда взять нужный репозиторий;
-
возможность выстраивать сложные сценарии в юз кейсах за счет вызова сервисов в одной транзакции;
-
прекрасная интеграция с внедрением зависимостей в
FastAPI.
Простое использование сессий избавило бы от необходимости аккумулировать репозитории в менеджере транзакций, создавать свой контекстный менеджер для транзакции, однако превратило бы сервисы и юз кейсы в большое полотно из простейших запросов. Для банального CRUD может быть и не страшно, но в моей разработке встречались случаи, когда надо было создать запись о платеже, внутренний запрос на членство в организации, сделать несколько проверок по базе, затем создать сессию оплаты, и обновить запись о платеже данными созданной платежной сессии.
Естественно, я к этому пришел не сразу. Пришлось посидеть и подумать, разобрать где и как у меня живет сессия, где закрывается старая, где открывается новая. Теперь я использую такой менеджер транзакций на постоянной основе, а в сочетании с CRUD-миксинами это ускоряет мою разработку, позволяя избавляться от дублирования кода и, в целом, писать более декларативный код.
Сортировка, фильтрация и пагинация
В принципе, реализация фильтрации, сортировки и пагинации — это невероятно хорошая и полезная практика для разработчика. В Django мы можем использовать эти вещи из коробки даже не задумываясь о том, как это реализовано, зачем оно нужно и как лучше передавать/принимать query-параметры для выполнения простых сортировок. Приводить свое полотно кода с реализацией условий сортировки и фильтрации не буду, это не тема статьи, зато немного поговорим про мой метод list, который принимает фильтры, сортировки, limit и offset.
Так как я разрабатывал систему с полноценной CMS, то работа с данными системы требовала гибкости — поиск по различным критериям, сортировки и пагинация. Всю эту логику я вынес в стандартный метод репозитория — метод list, который выглядит следующим образом:
async def list( self, limit: int = None, offset: int = None, order_by: str = None, filters: dict[str, Any] = None, stmt=None ) -> tuple[Sequence[T], int]: if stmt is None: stmt = select(self.model) count_stmt = select(func.count()).select_from(self.model) if filters: try: conditions = build_conditions(self.model, filters) except ValueError as e: raise InvalidFilterError(f"Invalid filter for {self.model.__name__}. Error: {e}") stmt = stmt.filter(*conditions) count_stmt = count_stmt.filter(*conditions) if order_by is not None: for param in order_by.split(","): desc_order = param.startswith("-") field_name = param.strip("-") if not hasattr(self.model, param.strip("-")): raise InvalidOrderAttributeError(f"{self.model.__name__} doesn't have attribute <{param}>") field = getattr(self.model, field_name) stmt = stmt.order_by(desc(field) if desc_order else asc(field)) if limit is not None and offset is not None: stmt = stmt.offset(offset).limit(limit) stmt = stmt.filter_by(_deleted=False) count_stmt = count_stmt.filter_by(_deleted=False) data = (await self.session.execute(stmt)).scalars().all() count = (await self.session.execute(count_stmt)).scalar_one() return data, count
Метод возвращает список объектов, а также общее количество найденных записей, применяет limit и offset, фильтры и сортировку. В сочетании с общей схемой для пагинированного ответа метод list позволяет забыть о том, что надо писать логику пагинации. Также я заложил возможность полностью переопределить стандартный sql-запрос. Необходимо это для подгрузки связанных сущностей или применения каких-то более сложных фильтров. Единственное, что мне действительно часто приходится делать — определять query-параметры для фильтров, которые я хочу принимать в эндпоинтах.
Пример эндпоинта с пагинацией, сортировкой и фильтрацией:
@router.get("")async def get_news_paginated_counted( service: NewsServiceDep, params: PaginationParamsDep, ordering: OrderingParamsDep = None, filters: Annotated[NewsFilter, Depends()] = None,) -> PaginatedResponse[NewsSchema]: data, count = await service.list( order_by=ordering, filters=filters.model_dump(exclude_none=True), limit=params["limit"], offset=params["offset"], ) return PaginatedResponse( count=count, data=data, page=params["page"], page_size=params["page_size"], )
Схема ответа, параметры запроса и фильтры:
# Для разных эндпоинтов изменяются только фильтрыDataModel = TypeVar("DataModel")class PaginatedResponse(BaseModel, Generic[DataModel]): count: int page: int page_size: int data: list[DataModel]def get_pagination_params( page: int = Query(1, ge=1, description="Page number"), page_size: int = Query(25, ge=1, le=100, description="Page size"),) -> dict: """returns limit, and offset page_size, page_size * (page - 1)""" return { "limit": page_size, "offset": page_size * (page - 1), "page": page, "page_size": page_size, }PaginationParamsDep = Annotated[dict[str, int], Depends(get_pagination_params)]OrderingParamsDep = Annotated[str | None, Query(description="Sorting parameters")]class NewsFilter(BaseModel): title__startswith: Annotated[str | None, Query(description="...")] = None is_published: Annotated[bool | None, Query(description="...")] = None created_at__gte: Annotated[datetime | None, Query(description="...")] = None created_at__lte: Annotated[datetime | None, Query(description="...")] = None updated_at__gte: Annotated[datetime | None, Query(description="...")] = None updated_at__lte: Annotated[datetime | None, Query(description="...")] = None
Решение достаточно простое, но именно такие небольшие абстракции экономят мне много времени при написании однообразных эндпоинтов. Стоит также отметить, что это решение далеко не идеально и, вероятнее всего, получит развитие или рефактор.
Soft delete в системе
В какой-то момент заказчик попросил функциональность архивации статей на сайте. Казалось бы — добавил boolean с названием _deleted и забыл. Но вскоре выяснилось, что архивация понадобилась и для других сущностей, а потом еще и как защита от дурака. Теперь все запросы
в системе знают про это boolean поле. Вроде бы подход простой, накинул фильтры в методах репозитория и забыл, но для разных систем это работает по-разному.
В одном моем проекте практически нет иерархии связей. Сущности не принадлежат одна другой, что делает soft delete безопасным, незаметным для пользователя и администратора сайта.
Но если рассмотреть систему даже с простой иерархией, то ситуация становится чуть менее радужной.
Допустим, есть образовательная система университета: факультеты, кафедры, группы, дисциплины. На факультете есть кафедры, на кафедрах группы, у групп — свои дисциплины, которые они изучают.
Что произойдет с группами при обычном удалении факультета из системы? При физическом удалении факультета база данных хотя бы может защитить нас ограничением внешнего ключа: например, при RESTRICT или NO ACTION удаление родительской записи будет отклонено. При soft delete никакого нарушения внешнего ключа не происходит — с точки зрения БД факультет продолжает существовать. Более того, пользователи системы будут также иметь доступ к кафедрам этого факультета, группам и так далее, просто они больше не увидят удаленный факультет. Это явно не то, что мы с вами хотим при разработке подобной системы, поэтому soft-delete накладывает свои ограничения. Решение проблемы можно выбирать из нескольких вариантов:
-
Каскадно удалять связанные записи без предупреждения;
-
Предупреждать пользователя о том, какие связанные сущности также будут удалены;
-
Запрещать удаление родительской сущности, пока пользователь не избавится от связанных записей.
В первом случае мы получаем неинтуитивное поведение системы — одно действие вызывает какие-то сайд-эффекты, это плохо с точки зрения UX, действия пользователя не должны иметь неочевидных побочных эффектов. Второй вариант звучит получше. Мы оставляем решение о каскадном удалении всех записей за пользователем, но всё равно сохраняем высокий риск массового удаления данных. Третий вариант самый безопасный — сохраняется высокая целостность данных, нет сайд-эффектов, однако это может банально замедлять работу администратора, а также требует хороших сообщений об ошибках.
В зрелой production-системе, в разработке которой я участвовал, я встречал именно третий подход. В одном из своих проектов реализовал то же самое, однако долго выбирал между подходами. Есть еще интересные подходы к soft delete, но я решил разобрать самые базовые.
Структура и архитектура FastAPI приложения
В прошлой своей статье я предлагал очень простую структуру, которая сейчас больше похожа на описание одного домена приложения:
Основное приложение находится в директории app/, где расположены папки:
models — модели SQLAlchemy;
api — все, что связано с роутингом;
utils — какие-то вспомогательные функции;
entities — pydantic модели.
Сейчас я бы разделил структуру проекта внутри папки /app (или /src, кому как удобнее) на 3 составляющие — /core, /domains и main.py. Почему так?
В core очень удобно вынести всякие конфигурации: база данных, файловое хранилище, утилиты, абстракции, какие-то общие для приложения исключения, Redis-client, криптография и так далее. Все, что определяется один раз в приложении и используется в доменах без дополнительных действий.
В domains находится код конкретных бизнес-доменов: бизнес-логика, сущности (они же модели), роуты, юз кейсы, схемы. Структура домена ограничена вашей фантазией и эффективностью ваших решений. Самое важное, что я хотел бы отметить — соблюдение потока зависимостей в приложении. Домены не должны быть сильно связаны, но связанность между ними практически неизбежна.
В моей схеме каждый следующий уровень собирает и использует возможности нижележащего, при этом нижние уровни ничего не должны знать о верхних. Это не обязательно про структуру файлов и папок, скорее про то, как работает приложение изнутри, кто что вызывает, кто кого реализует и тому подобное. Направление зависимостей в моем приложении выглядит примерно так:
Хочу отметить, что это не вычитанная в учебнике или статье архитектура. Что-то я взял из обучающих видео на YouTube, что-то где-то вычитал, а какие-то решения принимал самостоятельно или советовался с нейросетями. Самое важное здесь то, что такая архитектура достаточно хорошо работает, нормально тестируется и безболезненно расширяется.
Ну и последний структурный слой — main.py. Он собирает в себе все роуты всех доменов и запускает приложение. При этом он ничего не знает про сервисы, юз кейсы, модели, а они, в свою очередь, ничего не знают о нем, поэтому можно сказать, что это просто точка входа, которую можно будет достаточно легко изменить при необходимости.
Заключение
В заключение хочу сказать, что развитие системы, рефактор кода, архитектурные проблемы и их решение — это обязательные этапы в жизненном цикле системы. Приложение растет, кодовая база растет, появляются новые требования, а решения, которые прекрасно работали год назад, в какой-то момент начинают мешать дальнейшей разработке.
За последние полтора года мое представление о структуре FastAPI-приложения достаточно сильно изменилось. При этом я не могу сказать, что архитектура стала сложнее просто ради сложности. Большинство описанных в статье решений появились в тот момент, когда одна и та же проблема начинала повторяться достаточно часто: надоело вспоминать команды — появился скрипт, стало много однообразных запросов — появился общий list, понадобилось управлять несколькими действиями в одной транзакции — появился Transaction Manager, а рост количества бизнес-логики заставил четче определить границы между репозиториями, сервисами и юз кейсами.
Наверное, это и есть главный вывод, которым я хотел поделиться. Не стоит пытаться заранее построить идеальную архитектуру и предусмотреть все возможные сценарии развития проекта. Вряд ли это вообще возможно. Гораздо важнее понимать границы ответственности компонентов системы, следить за направлением зависимостей и не бояться менять собственные решения, когда они перестают решать поставленные перед ними задачи.
ссылка на оригинал статьи https://habr.com/ru/articles/1073130/