Привет! Меня зовут Влад и занимаюсь Python backend-разработкой. Довольно долго я работал над большим продуктом, который объединял несколько команд разработки. В нем было много микросервисов, базовые фичи кочевали из одного в другой, и часто разработчики делали одни и те же инфраструктурные компоненты по-разному. А когда переходили с одного микросервиса на другой, им приходилось долго осмыслять кодовую базу нового решения.
Код полнился ошибками, а разработчики тратили время на их отладку и исправление. Так мы вживую воплотили печальную статистику: разработчики ПО в среднем тратят на написание кода лишь 52 минуты в день, остальное — исправление ошибок и другие сопутствующие задачи.
Поэтому мы собрали небольшую команду разработчиков и вместе сделали шаблонизатор. Если проблема знакома, читайте дальше — расскажу, как он работает, поделюсь кодом и советами о том, как его эффективно применять.
Выбор, как стандартизировать разработку, несложный: опытные разработчики уже прошли через множество фреймворков и библиотек и сформировали рекомендации. Но когда речь заходит о том, как именно писать код, ситуация меняется.
Часто в ход идут сниппеты — готовые кусочки кода. Они позволяют быстро решать задачи, но в долгосрочной перспективе приводят к проблемам: теряется целостность архитектуры, появляются ошибки совместимости, становится сложно поддерживать структуру, проект может стать уязвимым с точки зрения безопасности.
Поэтому мы посмотрели в сторону шаблонизации. Шаблоны — это как рельсы для разработки. Они снимают с команды необходимость каждый раз принимать базовые решения и позволяют сосредоточиться на главном — создании бизнес-ценности. А реализация бизнес-логики приложений — это основной источник дохода компании.
По итогу шаблон помог нам сэкономить время на решении повторяющихся проблем, например, по подключению к базам данных, работе и интеграции с брокерами сообщений, реализации транзакций.
Конечно, этот проект не универсальный, мы делали его исходя из наших потребностей. Основная идея — собрать все лучшие практики написания кода и ведения проектов, которые накопили наши backend-специалисты. Мы собрали пул технологий, который будем использовать из проекта в проект, а еще развиваем сообщество разработчиков и продолжаем совершенствовать шаблонизатор.
Ознакомиться с шаблонизатором можно по ссылке на github проекта.
Описание шаблонизатора
Наш шаблонизатор — это набор стандартных решений для веб-сервиса, написанного с помощью FastAPI. Мы выбрали FastAPI, потому что он популярный, быстрый и простой — на нем мы делаем все наши новые проекты и уже накопили экспертизу по работе с этим фреймворком.
Шаблонизатор закрывает проблемы с типовыми задачами, например, с инфраструктурой проекта или его архитектурой. Достаточно просто выбрать, что должен уметь будущий шаблон, и у вас уже есть «каркас» будущего приложения — с документацией, готовой архитектурой проекта и стандартными фичами. А также с инструментами для локального развертывания — готовыми переменными окружения и docker-compose.
Есть много инструментов, способных генерировать шаблоны кода, но мы выбрали Cookiecutter. В нем достаточно простое API, можно быстро начать использование, есть подробная документация и большое сообщество — это помогает быстро закрывать вопросы, которые возникают при использовании.
Зачем делать свою реализацию
На просторах github можно встретить множество разных реализаций шаблонов с Cookiecutter для разработки сервисов на Python. Но у нас в компании уже есть свои стандарты и подходы к созданию продуктов, и среди готовых решений мы не нашли то, что подошло бы нам. Поэтому было решено сделать шаблонизатор «с нуля» — мы давно разрабатываем микросервисы и уже знаем, к какой архитектуре стремимся. Однако в будущем мы собираемся добавить новые для нас библиотеки и фичи, потому что тоже хотим вносить свой вклад в Open-source разработку.
Работа с шаблонизатором
Работа с шаблонизатором предельно проста. Для начала установите библиотеку Cookiecutter в виртуальном окружении или глобально.
Далее появляется выбор:
клонировать репозиторий с шаблонизатором на локальную машину и запустить генерацию шаблона с помощью команды
cookiecutter . --output-dir «new-project»
запустить генерацию прямо из удаленного репозитория с помощью команды
cookiecutter https://github.com/vladushka2000/PythonProjectTemplates
После начала инициализации проекта можно выбрать фичи, которыми должно обладать будущее приложение.

Пока что можно создать приложение, которое будет обладать следующими возможностями:
WSGI и ASGI-сервер;
асинхронная и синхронная работа с Postgres с помощью ORM SQLAlchemy и с помощью «сырых» запросов;
асинхронное и синхронное кеширование с помощью Valkey;
JSON-логгер;
Pre-commit;
автоматическая генерация docker-compose;
асинхронная работа с Kafka;
асинхронная работа с RabbitMQ;
интеграция с системами по REST API.
Когда вы выберете необходимые библиотеки, шаблонизатор сгенерирует «скелет» проекта с настроенными подключениями к внешним ресурсам и стандартными настройками для конфига. Конфиги вынесены в отдельные файлы в соответствии с библиотекой или какой-то отдельной фичей приложения, которую оно реализовывает.
Вместе с проектом также шаблонизатор создаст docker-файл, docker-compose и файл с переменными окружения для быстрого и удобного запуска проекта локально.
Для любого серьезного проекта документация – это must-have. Поэтому мы постарались полностью покрыть код комментариями. А еще реализовали функционал для генерации md-файлов с принципами и примерами работы: клиенты для подключения к брокерам, работа с UOW и т.д. Это позволяет быстрее погрузить пользователя в работу с готовым шаблоном и подстроить решение под себя.
Структура проекта
Основа проекта – это упрощенная чистая архитектура, в которой мы выделили всего 3 слоя.
Frameworks and drivers — это слой, который взаимодействует с внешним миром. Он включает в себя всё, что связано с инфраструктурой, фреймворками, библиотеками и внешними системами. Отвечает за взаимодействие с пользователями, другими сервисами, а также передачу данных во внутренние слои проекта.
Application Business Rules — этот слой содержит логику, специфичную для конкретного приложения. Он определяет, как данные обрабатываются и как выполняются сценарии приложения. Реализует основные функции приложения, но не зависит от внешних систем или инфраструктуры.Enterprise Business Rules — это внутренний слой, который содержит основную бизнес-логику проекта, его цель вне зависимости от конкретного приложения. Слой включает ключевые сущности и бизнес-правила, общие для всей организации, которые не зависят от внешних факторов.
По этим принципам приложение делится на следующие директории.

К слою Frameworks and drivers относятся директории:
brokers, в которой содержатся классы для подключения к брокерам сообщений;
repositories и storage, где реализованы объекты репозиториев и подключений, взаимодействующих с данными внешних систем;
web, которая содержит логику взаимодействия с системой при помощи API.
Слой Application business rules реализуется в директориях services, tools, uows и models/dto. В них находятся сервисы, утилиты, UOW и DTO, необходимые для корректной работы всего приложения.
Рассмотрим, как простой поток данных проходит по каждому из слоев.

Сначала данные приходят в слой Frameworks and drivers с помощью REST API — это наша директория web. Чтобы передать их дальше по слоям независимо друг от друга, создаются модели DTO, которым мы и будем пользоваться позже.
Полученные по API данные отправляются в слой Application business rules, в директорию services. Именно здесь задаются основные правила работы приложения.
Далее мы можем отправить данные в слой Frameworks and drivers, чтобы, например, получить информацию из базы данных, отправить сообщение в брокер и т.д.
Когда мы уже имеем все данные, можно собирать нашу доменную модель. Она импортируется из слоя Enterprise business rules и выполняет основную бизнес-логику сервиса.
Как только бизнес-логика была выполнена, пользователь может получить ответ об успешности операции. Для этого мы возвращаем необходимый DTO из слоя сервиса в слой API.
Мы исходим из принципа, что слои приложения должны быть максимально независимы друг от друга. Для этого в ход идет общение слоев с помощью отдельных объектов DTO, внедрение зависимостей и использование интерфейсов. Мы рекомендуем разбивать приложение по слоям, и шаблонизатор изначально планировался именно под такую организацию кода. Однако вы можете отойти от этих правил и подогнать архитектуру под свои потребности.
Концепция прокси-объектов
По всему шаблону тянется концепция, которая позволяет контролировать доступ к элементам систем с помощью шаблона проектирования «Прокси». Чтобы его реализовать, заводим в проекте следующий интерфейс:
import abc
class ConnectionProxy(abc.ABC):
"""
Proxy для подключения к удаленным ресурсам
"""
@abc.abstractmethod
async def connect(self, *args, **kwargs) -> any:
"""
Подключиться к ресурсу
"""
raise NotImplementedError
@abc.abstractmethod
async def disconnect(self, *args, **kwargs) -> any:
"""
Отключиться от ресурса
"""
raise NotImplementedError
На практике оказалось, что лишь два основных метода для подключения и отключения от ресурса покрывают все основные сценарии использования.
Как появилась идея завести абстракцию над подключением к другим ресурсам?
Изначально в шаблонизаторе была проблема: продюсер и консьюмер сообщений в RabbitMQ неэкономно использовали объект соединения, каждый имел свое собственное. Объект соединения для RabbitMQ поддерживает мультиплексирование, которое реализовано через объекты каналов. Гораздо логичнее было бы использовать один объект соединения на всю группу консьюмеров/продюсеров, которые семантически объединены по какому-то признаку.
Конечно, для этого можно использовать пул соединений. Но мы решили полностью контролировать процесс инициализации соединения с помощью прокси-объектов, которые отключались бы от брокера явным образом. Это дает разработчику полное управление временем жизни соединения, а также дополнительный контроль над тем, кто это соединение может использовать. Одним из вызовов оставался тот факт, что этот прокси-объект должен быть достаточно прост в использовании, поэтому мы сократили интерфейс настолько, насколько это возможно.
После создания прокси-объекта для RabbitMQ мы увидели, насколько удобно было пользоваться такой оберткой над соединением. Мы действительно могли настраивать его под себя, а простой интерфейс скрывал бы всю сложную логику. Так мы решили придерживаться того же самого подхода к работе и с другими внешними системами. Концепция прокси-объектов коснулась работы с Kafka, Postgres, системами, работающим по правилам REST и Valkey.
Для примера разберем реализацию для RabbitMQ.
class InstanceConnectionBase(base_proxy.ConnectionProxy):
"""
Базовый класс прокси-подключения для разных сущностей с реализацией aio_pika
"""
_connection: aio_pika.abc.AbstractRobustConnection | None = None
_connection_users = set()
_connection_lock = asyncio.Lock()
@classmethod
async def _set_connection(cls) -> None:
"""
Установить соединение
"""
cls._connection = await aio_pika.connect_robust(str(config.rabbit_mq_dsn))
@classmethod
def _add_connection_user(
cls, new_user: base_message_broker.BaseConsumer | base_message_broker.BaseProducer
) -> None:
"""
Установить число объектов, которые используют соединение
:param new_user: новый пользователь
"""
cls._connection_users.add(new_user)
@classmethod
def _delete_connection_user(
cls, user: base_message_broker.BaseConsumer | base_message_broker.BaseProducer
) -> None:
"""
Установить число объектов, которые используют соединение
:param user: пользователь
"""
cls._connection_users.remove(user)
async def connect(
self, user: base_message_broker.BaseConsumer | base_message_broker.BaseProducer
) -> aio_pika.abc.AbstractRobustConnection:
"""
Подключиться к брокеру
:param user: пользователь соединения
:return: объект соединения
"""
self._add_connection_user(user)
async with self._connection_lock:
if self._connection is None:
await self._set_connection()
return self._connection
async def disconnect(
self, user: base_message_broker.BaseConsumer | base_message_broker.BaseProducer
) -> None:
"""
Отключиться от брокера
:param user: пользователь соединения
"""
if self._connection is None:
raise ValueError("Объект соединения не инициализирован")
self._delete_connection_user(user)
if len(self._connection_users) == 0:
await self._connection.close()
self._connection = None
Рассмотрим алгоритм работы публичных методов прокси-соединения.
Сначала мы добавляем в переменную класса _connection_users, которая является множеством, объект получателя или отправителя сообщений. Это нужно для отслеживания активных уникальных пользователей соединения.
Проверяем, инициализирован ли объект соединения для класса. Если нет, то в переменную класса _connection записываем объект соединения. После этого объект соединения отдается пользователю. Таким образом, даже при множественном вызове метода connect() мы будем получать один и тот же объект соединения. Получается некий Singleton — и им могут пользоваться различные классы, которым нужно подключение к удаленному ресурсу.
Чтобы отключиться от внешнего ресурса, в прокси-объект введена проверка на количество пользователей соединения. Сначала мы удаляем текущего пользователя из множества, которое хранится в переменной _connection_users, а затем смотрим, остались ли еще пользователи в этом множестве. Если их не осталось, то мы можем закрывать соединение.
Как видно, логика прокси-соединения несложная, однако дает хороший контроль над переиспользованием текущих соединений. В связке с паттернами «Репозиторий» и UOW этот объект позволит удобно управлять временем жизни подключения к внешним объектам.
На самом деле прокси может использоваться не только для реализации соединений, но и, например, как обычная обертка над любыми объектами. Так, в шаблонизаторе реализована возможность делать «сырые» запросы к базе данных Postgres с помощью клиентских и серверных курсоров в синхронном и асинхронном режиме. В некоторых местах прокси-объекты мимикрируют под курсоры, но на самом деле это обычные объекты соединения. Это сделано, чтобы обходить ограничения библиотек и соблюдать единый интерфейс работы со всеми объектами.
Репозитории и Unit of Work
Паттерн «Репозиторий» — это слой доступа к данным, абстракция, которая позволяет работать с хранилищами вне зависимости от их типа. Для этого мы ввели следующий интерфейс, который повторяет операции CRUD и расширяет их методы получения списка объектов.
import abc
from typing import Iterable
class BaseRepository(abc.ABC):
"""
Базовый класс репозитория
"""
@abc.abstractmethod
def create(self, *args, **kwargs) -> any:
"""
Создать запись
"""
raise NotImplementedError
@abc.abstractmethod
def retrieve(self, *args, **kwargs) -> any:
"""
Получить запись
"""
raise NotImplementedError
@abc.abstractmethod
def list(self, *args, **kwargs) -> Iterable[any]:
"""
Получить список записей
"""
raise NotImplementedError
@abc.abstractmethod
def update(self, *args, **kwargs) -> any:
"""
Обновить запись
"""
raise NotImplementedError
@abc.abstractmethod
def delete(self, *args, **kwargs) -> any:
"""
Удалить запись
"""
raise NotImplementedError
Предполагается, что если не хватит методов в интерфейсе, будет создан новый репозиторий на основе того интерфейса, который реализует недостающий функционал. И наоборот — когда, например, необходимо реализовать лишь один метод из имеющихся, нарушается принцип Interface segregation. Но этот факт намеренно игнорируется, потому что мы разрабатывали интерфейс репозитория с оглядкой на универсальность конечного решения — чтобы его было удобно использовать в дальнейшем.
В шаблоне репозитории имеют доступ к прокси-соединениям. При инициализации они передаются как поле объекта. Это позволяет получать данные через соединение, но лишает пользователя удобных способов управления транзакцией и освобождением ресурсов. Для этого рекомендуем использовать UOW вместе с репозиториями, чтобы избежать утечек памяти. UOW реализован как контекстный менеджер, что делает его удобным для добавления логики подключения и отключения от внешних ресурсов.
from __future__ import annotations # noqa
import abc
from interfaces import base_repository
class BaseSyncUOW(abc.ABC):
"""
Абстрактный класс синхронного UOW
"""
def __init__(self, *args, **kwargs) -> None:
"""
Инициализировать переменные
"""
...
def __enter__(self, *args, **kwargs) -> BaseSyncUOW:
"""
Войти в контекстный менеджер
"""
return self
def __exit__(self, *args, **kwargs) -> None:
"""
Выйти из контекстного менеджера
"""
self.rollback()
@abc.abstractmethod
def commit(self) -> None:
"""
Сделать коммит изменений
"""
raise NotImplementedError
@abc.abstractmethod
def rollback(self) -> None:
"""
Сделать откат изменений
"""
raise NotImplementedError
Как видно из кода, подключение идет в методе __enter__ (**__aenter__**для асинхронной версии) и освобождение ресурсов в __exit__ (**__aexit__**для асинхронной версии). Также добавили явные методы для применения и отката изменений, так как явное лучше неявного. Так пользователь получает прозрачную работу с ресурсами для подключения к внешним источникам.
Сам же репозиторий может передаваться в инициализатор объекта UOW. Например, так это было реализовано для UOW, который работает с репозиториями asyncpg.
class AsyncpgUOW(base_uow.BaseAsyncUOW):
"""
Асинхронный UOW для работы с асинхронными Psycopg-репозиториями
"""
def __init__(self, repository: asyncpg_repository.AsyncpgRepository) -> None:
"""
Инициализировать переменные
:param repository: асинхронный репозиторий Psycopg
"""
self.repository = repository
self._cursor_proxy: base_postgres_cursor_proxy.BaseAsyncpgCursorProxy | None = None
self._transaction: asyncpg.transaction.Transaction | None = None
self._is_transaction_commited = False
super().__init__()
# Остальная часть кода
Такая гибкость позволяет добавлять логику для репозитория при работе с транзакциями.
Слой сервиса
По чистой архитектуре приложение должно иметь слой пользовательских сценариев и слой сервиса. Пользовательские сценарии отвечают за описание конкретного способа использования системы — что должно быть сделано для достижения цели. А реализацией этой цели уже занимается слой сервиса. Однако в шаблонизаторе мы немного изменили это и для простоты объединили эти два слоя в один — слой сервиса.
Объект этого слоя не имеет интерфейса, он может реализовывать любые методы. Предполагается, что пользователь называет сервис как обобщающую сущность для какой-то группы действий. Например, AuthService — для управления аутентификацией, CartService — для управления корзиной покупок пользователя и т.д. Методы сервиса будут конкретными пользовательскими сценариями. Например, auth() для сервиса аутентификации, add_item() для сервиса управления корзиной покупок пользователя. Конкретные методы уже могут вызывать репозитории для получения данных, создавать доменные модели и реализовывать логику самого приложения.
Такое упрощение позволяет не «раздувать» объем кода дополнительными абстракциями, а объединить их, потому что они имеют схожую функциональность.
Внедрение зависимостей
Для объединения слоев и уменьшения зацепления наш шаблон реализует DI-контейнеры во всех частях приложения. Это избавляет пользователя от слишком сильной зависимости разных слоев друг от друга и позволяет выстроить поток данных, который идет от внешнего слоя фреймворков и API к внутреннему — слою доменной модели.
Стандартные реализации DI-контейнеров уже «вшиты» в шаблон. В них же можно посмотреть логику создания объектов. Например, разберем контейнер для синхронной работы с «сырыми» запросами к Postgres.
class PsycopgSyncContainer(containers.DeclarativeContainer):
"""
DI-контейнер с провайдерами для работы с БД Postgres через psycopg
"""
# указать связанные модули
wiring_config = containers.WiringConfiguration(modules=None)
connection_pool_factory = providers.Singleton(
psycopg_connection_pool_factory.PsycopgPoolFactory,
config.postgres_dsn,
config.connection_pool_size,
)
cursor_type = providers.AbstractFactory(cursor_proxy.BasePsycopgCursorProxy)
connection_proxy = providers.Factory(
connection_proxy.PsycopgConnectionProxy, connection_pool_factory, cursor_type
)
# Добавить провайдеры конкретных реализаций репозиториев
psycopg_repository = providers.Factory(
psycopg_repository.PsycopgSyncRepository, connection_proxy
)
# Добавить провайдеры конкретных реализаций UOW
psycopg_uow = providers.Factory(psycopg_uow.PsycopgSyncUOW, psycopg_repository)
Здесь сразу инициализируются все необходимые компоненты: объект прокси-соединения, пул соединений, тип курсора, репозиторий и UOW для управления им.
Контракты взаимодействия с брокерами сообщений
Независимо от того, используется ли Kafka или RabbitMQ, в приложении установлен контракт взаимодействия системы и брокера сообщений. Он выглядит следующим образом:
import datetime
import uuid
from pydantic import Field
from interfaces import base_dto
class BrokerMessageDTO(base_dto.BaseDTO):
"""
DTO, содержащий сообщение для брокера
"""
id: uuid.UUID = Field(
description="Идентификатор сообщения", default_factory=lambda: uuid.uuid4()
)
body: dict = Field(description="Тело сообщения")
date: datetime.datetime = Field(
description="Дата создания сообщения",
default_factory=lambda: datetime.datetime.now(),
)
Сообщение для брокера очень минималистичное: уникальный идентификатор в виде UUID, тело сообщения (сериализуемый в json словарь) и дата отправки.
Остановимся на теле сообщения. Предполагается, что пользователи сами будут задавать дополнительные pydantic-модели для тела сообщения. Например, так выглядит сообщение-команда на расчет, которую мы передадим в воображаемый расчетный сервис:
import datetime
import uuid
from pydantic import Field
from interfaces import base_dto
from tools import enums
class EDATypeDTO(base_dto.BaseDTO):
"""
DTO, содержащий информацию о типе события и его свойства
"""
event_type: enums.CommandType = Field(description="Тип события EDA")
object_id: uuid.UUID = Field(description="Идентификатор объекта расчета")
message: str = Field(description="Описание события")
calc_date: datetime.datetime = Field(description="Дата замеров для расчета")
И вот так выглядит код для формирования финального сообщения, которое будет отправлено в брокер сообщений:
message = broker_message_dto.BrokerMessageDTO(
body=dict(
eda_dto.EDATypeDTO(
event_type=enums.CommandType.START_CALC.value,
cluster_id=cluster_id,
message="",
calc_date=dt
)
)
)
Достаточно просто и лаконично. Таким образом можно делать вложенные сообщения и считывать их в другом сервисе. Главное здесь - это соблюдение контракта в системе.
Пример использования
Для наглядности и быстрого погружения мы придумали простой пример.
Пользователь отправляет запрос в сервис в bff на расчет, который требует значительных ресурсов процессора.
Сервис bff формирует сообщение на расчет и отправляет его в брокер RabbitMQ в очередь commands.
Эту очередь читает воркер calculator worker. При получении команды он делает запрос в базу данных в соответствии с полученным сообщением и начинает расчет. Закончив расчет, он формирует новое сообщение с результатами и отправляет его в очередь events, которая далее может читаться другим сервисом, но это уже за рамками нашего примера.
Если обобщать архитектуру системы, то можно нарисовать простую схему.

В примере мы смогли реализовать полноценную доменную модель для воркера, настроить интеграцию сервисов с помощью брокера RabbitMQ, сделать ci — и все это благодаря одному шаблону. Мы сэкономили уйму времени на реализации стандартных вещей и сразу приступили к бизнес-логике.
Что планируем сделать в будущем
Шаблонизатор постоянно улучшается — мы внедряем его для локальных inner-source проектов и активно собираем отзывы о работе с ним.
Одним из больших шагов станет внедрение конфигурации, которая позволит подстраивать шаблон под конкретный тип микросервиса. Например, под bff, воркер или просто вспомогательный микросервис, время жизни которого было бы ограничено выполняемым скриптом.
Также работаем над внедрением библиотеки тестирования pytest, которая бы сразу задавала шаблоны для написания всех видов тестов.
Как шаблонизатор помогает бизнесу
Сокращает time-to-market, новые сервисы будут создаваться быстрее.
Сокращает время на онбординг новых сотрудников: мы вводим стандарты написания кода и best practices внутри отдела разработки, что снижает порог входа в новый проект для сотрудников, которые перемещаются между проектами.
Минимизирует рутину и уменьшает вероятность ошибок: базовые функции для старта нового микросервиса уже реализованы и настроены «из коробки», что оставляет только фокус на бизнес-логике.
Заключение
Шаблонизатор — это объединяющее звено всех наших практик разработки, которые мы собирали у себя в компании. Мы попытались сделать его максимально универсальным, при этом сохраняя простоту API.
Конечно, он находится в стадии активной разработки и еще требует доработок и улучшений: нужны фиксы существующих проблем, новые фичи и библиотеки. Но уже сейчас можно однозначно сказать, что шаблон помогает в реализации простых CRUD-сервисов.
Мы стараемся продвигать наше решение и призываем вас поучаствовать в разработке проекта — любой может оставить свой Issue или открыть Pull Request на github.
Мы верим, что подобные проекты могут развивать сообщество backend-разработчиков на Python, привносить свежий взгляд на решение старых проблем, а главное — вдохновлять на создание собственных решений.