Наш архитектурный подход к Python приложениям

Мы долгие годы писали сервисы, исходя из каких-то своих внутренних ощущений правильности их написания. Но синхронизироваться по хорошим практикам в разных командах бывает довольно сложно и часто хорошие практики не выходили за рамки одной команды, а такого хотелось бы избежать. Поэтому мы решили объединить все хорошие по нашему мнению практики в единый справочник. Этот справочник получил название «Архитектурный гайд». Про него и поговорим в данной статье.

Сразу хотим предупредить! Несмотря на то, что эти рекомендации применяются в наших командах и кажутся нам довольно полезными, они вовсе не являются единственно верными. Мы никого не призываем отказываться от своих архитектур и не говорим, что наши рекомендации лучше остальных. Воспринимайте данный материал как один из возможных справочников по проектированию и разработке сервисов на Python.

К сожалению, в этой статье физически не получится рассмотреть абсолютно все инструменты, которые мы применяем. Если вы хотите узнать обо всех применяемых нами инструментах, можете прочитать данный доклад ^[1].

Код стайл

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

• Ruff — супер-быстрый линтер, который значительно превосходит по скорости и количеству правил остальные решения.

• Mypy — тайпчекер, который на момент написания статьи содержит в себе наибольшее количество правил и отлично работает.

• Hadolint — линтер для докер-файлов помогает придерживаться best practices при их написании.

• kube-score — линтер для k8s манифестов тоже для best practices.

При написании кода мы также придерживаемся принципа разделения ответственности. Разделяем функциональность объектов так, чтобы можно было четко ответить на вопрос: «Чем занимается данный объект?». Если в ответе получается несколько зон ответственности, то в 99% случаев такой объект должен быть разделен. Если упростить этот принцип до одного предложения, получится, что «Объект должен содержать минимально-необходимый набор методов для корректного выполнения своей функции».

Настройки

В нашей концепции у каждого сервиса собственный набор настроек, необходимых для правильной работы. Это могут быть пароли, имена пользователей, ссылки на сервисы и так далее. Все настройки мы храним в файле settings.py

Пример лейаута:

some-app/
└── some_app/
   ├── main.py
   └── settings.py

Пример файла:

class ServiceSettings(pydantic_settings.BaseSettings):
    service_name: str = "micro-service",
    service_description: str = "Micro service description"

    database_url: str = "your-database-url"
    database_read_timeout: int = 5

    redis_url: str = "your-redis-url"

    ...

    # используется, чтобы загружать переменные из файла .env
    model_config = pydantic_settings.SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        populate_by_name=True,
    )

settings = ServiceSettings()

Чтобы использовать эти настройки, достаточно импортировать их в другом модуле. В основе паттерна настроек лежит библиотека pydantic_settings, про нее можно прочитать здесь ^[2]

API

Мы используем два довольно популярных веб-фреймворка, которые предоставляют всё необходимое для создания API (Application Programming Interface): fastapi ^[3] или litestar ^[4].

Организация API в проекте

Мы размещаем API обработчики в модуле api и группируем их по протоколам/технологиям. Если обработчиков не слишком много и они без проблем влезают в один файл, то так и оставляем.

Пример лейаута:

some-app/
└── some_app/
   ├── __main__.py
   └── api/
       ├── __init__.py
       ├── rest.py
       ├── rpc.py
       └── soap.py

Если же обработчики не влезают в один файл и с ними неудобно работать, можно разделить api по доменам приложения, группируя в модули по протоколу/технологии. Тогда получится следующая структура:

some-app/
└── some_app/
   ├── __main__.py
   └── api/
       ├── rest/
          ├── __init__.py
          ├── products.py
          ├── users.py
          └── ...
       └── rpc/
          ├── __init__.py
          ├── file_handling.py
          ├── admin_operations.py
          └── ...

У нас в сообществе приняты следующие типы API:

• rest — CRUD для манипуляции сущностями приложения, который подходит для большинства рядовых задач. Например: GET /api/books/. В то же время все действия выполняются через http методы (в данном случае GET, POST, PUT, DELETE), а не содержатся в названии самой ручки. Более подробно про REST можно почитать здесь ^[5]

• rpc — используем для ручек, которые выполняют какие-то действия, поэтому их нельзя отнести к REST. Пути до таких обработчиков обязательно должны содержать в себе глагол. Например: /rpc/sort-binary-tree/, /rpc/send-file/, /rpc/upload-to-pypi/

Версионирование HTTP API

Мы стараемся избегать версионирования REST API, так как это приводит ко многим сложностям в поддержке такого API. Например, если вы делаете внутреннее приложение и пишете для него фронтенд (чаще всего это наш кейс) — версионирование вам не нужно. Но есть исключение, такое как отсутствие контроля над потребителем API (классический случай — мобильное приложение).

Самый же «рестовый» способ — версионировать accept заголовком. На этот случай мы используем библиотеку fast-version ^[6].

Альтернативно можно версионировать с помощью пути — /api/v1/, /api/v2/. Такой способ не особо вписывается в REST, так как REST — это архитектурный стиль api для манипуляции сущностями, а версия — не часть сущности. User — всегда user, а не userv1. Однако такой способ получил популярность, потому что удобен. Для него почти ничего не требуется.

Для себя мы решили, что версионирование делаем только через заголовки в случае очень сильно надобности.

Dependency injection

Мы придерживаемся DIP (Dependency Inversion Principle) из SOLID при написании сервисов. В этом нам помогает паттерн DI (Dependency Injection).

Однако полностью DIP мы не достигаем, так как мы завязываемся не на абстракции, а берем конкретные объекты. Это становится возможным, так как в Python утиная типизация. То есть мы можем «подпихнуть» любой объект, если он соблюдает требуемый интерфейс. Иными словами, DIP мы одновременно и соблюдаем, и нет.

Раньше для нас приоритетным пакетом был dependency-injector ^[7], но он долгое время не поддерживал Python 3.12 и мы перебрались на that-depends ^[8]. Этот пакет предлагает все те же концепции, что и dependency-injector, но при этом намного меньше и проще.

Есть и другие DI пакеты, которые позволяют делать примерно те же самые вещи. Например: dishka ^[9] и modern-di ^[10].

Работа с очередями

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

some-app/
└── some_app/
   ├── __init__.py
   ├── __main__.py
   └── consumers/
       ├── consumer_1.py
       └── consumer_2.py

За последний год большую популярность получила библиотека FastStream ^[11]. Она значительно упрощает создание консьюмеров и хорошо подходит нам, так как мы в основном используем Kafka и NATS. FastStream автоматически генерирует AsyncAPI документацию (AsyncAPI ^[12]) для наших консьюмеров. Это аналог Swagger ^[13] для HTTP API, но предназначенный для описания консьюмеров.

Вот небольшой пример работы с NATS-очередью с использованием FastStream:

from faststream import FastStream
from faststream.nats import NatsBroker

# Создаем брокер, который будет работать с инстансом NATS
broker = NatsBroker("nats://broker-host:4222")
app = FastStream(broker)

# Регистрируем обработчик в стиле FastAPI
@broker.subscriber("example.subject")
async def handle_message(message: str):
   print(f"Received message: {message}")
   # Ваша логика обработки сообщения


if __name__ == "__main__":
   app.run()

С FastStream мы тратим меньше сил на создание консьюмеров и можем больше сосредоточиться на логике приложения. Если ещё не пользовались, почитайте официальную документацию ^[14].

Конечно, учитывая, что мы работаем в банке, у нас есть много легаси-инфраструктуры, например, ActiveMQ Artemis. Для таких случаев мы разработали собственный пакет ^[15], совместимый с FastStream.

Сервисный слой

Сервисный слой — центр бизнес-логики приложения. Как правило, сервис — это класс, методы которого реализуют бизнес-сценарий.

Мы используем вот такой лейаут:

some-app/
└── some_app/
   ├── init.py
   ├── main.py
   └── services/
       ├── init.py
       ├── some_service_1.py
       └── some_service_2.py

При написании сервисов мы придерживаемся следующих правил:

1. Сервис может взаимодействовать с БД/кешами/очередями/сетью/любой другой инфраструктурой только через зависимости, например, через соответствующих клиентов или репозитории.

2. Только сервисы содержат бизнес-логику.

3. Сервисы могут использовать другие сервисы внутри себя, если это необходимо.

4. Сервис рекомендуется определять при помощи dataclasses со следующими параметрами:

• kw_only — параметры для dataclass передаются только через именованные параметры, не через последовательные. Это уменьшает вероятность неправильной передачи параметров.

• frozen — параметры передаются в процессе создания и не могут быть изменены в процессе работы. Это добавляет консистентности в работе сервиса, обеспечивая идентичную работу после создания.

• slots — генерирует slots у dataclass и таким образом ограничивает набор атрибутов у экземпляра класса. Это экономит использование памяти и ускоряет доступ к атрибутам.

Например, при создании и удалении пользователя нам необходимо отправлять соответствующее сообщение в какую-нибудь очередь в Kafka.

Вот этот код не самого лучшего качества:

@dataclasses.dataclass
class UserService:
    database_host: str
    database_port: str
    kafka_topic: str
    kafka_host: str

    async def create_user(self, user_data: UserData) -> None:
        database_connection = asyncpg.connect(
            self.database_host, 
            self.database_port
        )
        await database_connection.execute("INSERT INTO users ...")
        kafka_connection = aiokafka.connect(self.kafka_host)
        await kafka_connection.send(user_data, self.kafka_topic)

    async def delete_user(self, user_id: str) -> None:
        database_connection = asyncpg.connect(
            self.database_host, 
            self.database_port
        )
        await database_connection.execute("DELETE FROM users WHERE ...")
        kafka_connection = aiokafka.connect(self.kafka_host)
        await kafka_connection.send(user_data, self.kafka_topic)

Код не самый хороший потому, что:

• нарушен принцип единой ответственности из SOLID

• код тяжело переиспользовать 

• код тяжело воспринимать тому, кто его не писал

• код тяжело тестировать

• в определении dataclass не использованы: kw_only, frozen, slots

Такой код уже намного лучше:

@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
class UserService:
    user_repository: UserRepository
    kafka_producer: KafkaProducer

    async def create_user(self, user_data: UserData) -> None:
        await self.user_repository.create_user(user_data)
        await self.kafka_producer.send_message(user_data)

    async def delete_user(self, user_id: str) -> None:
        await self.user_repository.delete_user(user_id)
        await self.kafka_producer.send_message(user_id)

Этот пример лучше потому, что:

• код хорошо читается

• код можно переиспользовать, ведь вся логика взаимодействия с БД и очередью находится в отдельных объектах

• тестируемость кода заметно выше, так как можно легко тестировать методы зависимостей сервиса

• dataclass ведет себя намного более предсказуемо из-за kw_only, frozen, slots

Работа с базой данных

Когда у нас возникает необходимость в реляционной БД, мы используем связку SQLAlchemy ^[16] + Alembic ^[17] как самую популярную и проверенную временем.

Наш лейаут:

some-app/
└── some_app/
   ├── main.py
   └── database/
       ├── init.py
       ├── migrations/
           ├── versions/...
           ├── env.py
           └── script.py.mako
       ├── connection.py
       ├── models.py
       └── repositories.py

Всё, что связано с Alembic и миграциями хранится в database/migrations. 

• versions/ — здесь расположены сами файлы миграций Alembic

• env.py — используется для настройки Alembic (запуск/форматирование миграций/прочее)

• script.py.mako — Mako шаблон для генерации миграций через alembic revision --autogenerate

Что касается основных файлов на верхнем уровне:

• connection.py — инструментарий для создания соединений с базой. Управление сессиями, модификации движков SQLAlchemy и прочее.

• models.py — описание таблиц и их взаимоотношений в БД. Если моделей много — их можно разнести по разным файлам.

• repositories.py — набор репозиториев для работы с БД. Если моделей много — их можно разнести по разным файлам и положить рядом с моделью.

Репозитории

Репозитории содержат методы для общения с БД. Это могут быть как CRUD методы, так и комбинация CRUD вызовов внутри одного метода.

Мы разделяем репозитории по принципу единой ответственности. Одна таблица — один репозиторий. Для облегчения их создания мы используем библиотеку Advanced Alchemy ^[18]. Она предоставляет базовый класс со всеми необходимыми CRUD операциями над sqlalchemy моделями и много чего еще. Все, что нужно — просто указать модель, на основе которой следует создать репозиторий.
Вот пример репозитория на основе базового репозитория из этой библиотеки:

class PostRepository(SQLAlchemyAsyncRepository[Post]):
    """Repository for managing blog posts."""
    model_type = Post

    # уже реализовано в родительском классе репозитория
    async def add(...)
    async def add_many(...)
    async def list(...)
    async def update(...)
    async def delete(...)
    ...

На основе этой библиотеки можно создавать собственные базовые репозитории для дальнейшего переиспользования. Advanced Alchemy совместима как с litestar, так и с fastapi.

Интеграционный слой

Чтобы интегрироваться с любыми внешними источниками (Redis, S3, любой другой сервис), мы выделяем сущности для общения с ними в отдельный модуль и называем «клиентами».

Интеграционный слой состоит из клиентов, которые обращаются к другим сервисам или инфраструктуре. Клиенты принято размещать в папке external на корневом уровне или же в файлах сервисов, в которых они используются.

some-app/
└── some_app/
   ├── main.py
   └── external/
       ├── init.py
       ├── s3.py
       └── some_client.py

Простейший клиент может выглядеть так:

import dataclasses

@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
class HTTPClient:
    service_host: str

    def make_request(self, method: str, data: dict, ...) -> dict: ...

А чтобы такой клиент был надежным, мы добавляем в него resilience.

Resilience

При неудачном выполнении запроса его следует повторить. При помощи повторных запросов мы довольно дешево увеличиваем надежность системы. Для этой цели мы выбрали библиотеку stamina ^[19].

Пример resilient клиента с использованием stamina:

import stamina

class HTTPClient(...):
    @stamina.retry(on=CustomException, attempts=3, ...)
    def make_request(self, method: str, data: dict, ...) -> dict: ...

Есть и другие библиотеки, такие как: backoff ^[20] или tenacity ^[21]. Но stamina показалась нам самой подходящей.

Помимо повторных запросов, для предотвращения перегрузки системы при многочисленных ошибках можно установить Circuit Breaker. Он позволяет временно приостановить выполнение запросов к ресурсу, который регулярно возвращает ошибки, чтобы дать ему время на восстановление. Но если вы не видите реальной угрозы положить какую-то из систем избыточными запросами, можно обойтись и без Circuit breaker. Почитать про паттерн Circuit Breaker можно здесь ^[22].

Пример реализации клиента с Circuit Breaker на основе библиотеки circuitbreaker ^[23]:

from circuitbreaker import circuit

class HTTPClient(...):
    @circut(failure_threshold=10)
    def make_request(self, method: str, data: dict, ...) -> dict: ...

Тесты

Для написания тестов мы, как, наверное, и большинство, используем фреймворк Pytest ^[24]. Для себя мы выработали следующий лейаут:

some-app

├── some_app
├── api/
└── users.py
├── services/
└── comments_service.py
└── consumers/
└── posts_consumer.py
└── tests/
├── conftest.py
   ├── factory.py
├── helpers.py
   ├── api/
  └── test_users.py
├── services/
└── test_comments.py
└── consumers/
└── test_posts.py

По сути, мы копируем названия модулей и файлов, полностью повторяя структуру проекта, но к названиям файлов добавляем префикс “test_”, как того требует Pytest. А помимо файлов с тестами в тот же модуль кладем вспомогательные файлы:

• conftest.py — стандартный файл для фреймворка Pytest, там находятся фикстуры. Прочитать про него можно тут ^[25]. Он может находиться внутри любого другого модуля, содержащего тесты. Если мы понимаем, что при тесте сервисов или любой другой сущности у нас будут какие-то особенные фикстуры, относящиеся только к ним, то можно положить их рядом с тестами. Пример пути до такого файла: tests/[api | services | consumers | …]/conftest.py.

• factory.py — здесь хранятся объекты Factory, которые помогают при генерации фейковых данных для тестов. Раньше мы использовали faker или factory-boy, но перешли на более новую библиотеку Polyfactory ^[26]. 

• helpers.py — сюда мы складываем вспомогательные функции, необходимые для проведения тестирования и соблюдения принципа DRY.

Мы выработали для себя следующие рекомендации по написанию тестов:

Рассмотрим несколько примеров теста. Сначала не самый хороший тест:

def test_some_logic(my_service: MyService) -> None:
    test_value_1 = "test-value-1"
    test_result_1 = "test-result-1"
    received_result_1 = my_service.perform_some_logic(test_value_1)
    assert received_result_1 == test_result_1

    test_value_2 = "test-value-2"
    test_result_2 = "test-result-2"
    received_result_2 = my_service.perform_some_logic(test_value_2)
    assert received_result_2 == test_result_2

    test_value_3 = "test-value-3"
    test_result_3 = "test-result-3"
    received_result_3 = my_service.perform_some_logic(test_value_3)
    assert received_result_3 == test_result_3

И вот почему этот тест не самый хороший:

• Тест тяжело читать, ведь непонятно, что именно он проверяет.

• Тест не использует паттерн ААА.

• Тест не использует параметризацию.

Намного лучше так:

@pytest.mark.parametrize(
    ("test_value", "expected_result"),
    [
        ("test-value-1", "test-result-1"),
        ("test-value-2", "test-result-2"),
        ("test-value-3", "test-result-3"),
    ],
)
def test_some_logic(test_value: str, expected_result: str, my_service: MyService) -> None:
    assert my_service.perform_some_logic(test_value) == expected_result

Еще пример не самого хорошего теста.

def test_some_different_logic(my_service: MyService) -> None:
    test_value_1 = "value-first-test-func"
    test_result_1 = "result-first-test-func"
    received_result_1 = my_service.first_function_to_test(test_value_1)
    assert received_result_1 == test_result_1

    test_value_2 = "value-second-test-func"
    test_result_2 = "result-second-test-func"
    received_result_2 = my_service.second_function_to_test(test_value_2)
    assert received_result_2 == test_result_2

Он плохо написан потому, что:

• Не следует ААА.

• В одном тесте тестируются сразу две функции - first_function_to_test и second_function_to_test. Так быть не должно, тест обязан проверять что-то одно.

Намного лучше так:

def test_first_func(my_service: MyService) -> None:
    test_value = "value-first-test-func"
    test_result = "result-first-test-func"
    received_result = my_service.first_function_to_test(test_value)
    assert received_result == test_result

def test_second_func(my_service: MyService) -> None:
    test_value = "value-second-test-func"
    test_result = "result-second-test-func"
    received_result = my_service.second_function_to_test(test_value)
    assert received_result == test_result

Здесь все хорошо, каждый из тестов проверяет что-то одно.

Запуск тестов

Для себя мы решили запускать тесты через compose как локально, так и на стадии CI.
На то есть несколько причин:

• Контейнерное окружение дает лучшую изоляцию при тестировании.

• В пайплайнах тесты также гоняются через compose, поэтому локально мы получаем репрезентативность 1к1.

• Вся инфраструктура для интеграционных тестов поднимается в compose автоматически, что удобно.

Пример compose:

services:
 application:
   image: ... 
   depends:
	- migrations 
   ports:
     - 8000:8000
   environment:
     - ...

 database:
   image: postgres
   ports:
     - 5432:5432

 migrations:
   image: ...
   depends:
	- database 
   command: alembic upgrade head

Здесь application — наш веб-сервис. Помимо него можно увидеть базу данных и миграции. Миграции будут автоматически применяться перед каждым запуском.

У сервиса application не указана команда для запуска. Это связано с тем, что команда подставляется в момент запуска тестов. Вот такую команду мы выполняем, чтобы запустить тесты локально или на стадии CI:

docker compose run application pytest -sv -n 4 --cov=. --cov-report term-missing

Этой командой мы запускаем тесты в 4 процесса с помощью pytest-xdist ^[29], а также собираем отчет о тестировании при помощи pytest-cov ^[30].

Microbootstrap

В наших сервисах довольно много внимания уделяется observability. Поэтому, чтобы не таскать boilerplate из проекта в проект, мы создали пакет, который делает это за нас. Почитать про нее подробнее можно здесь ^[31].

С помощью microbootstrap можно создавать сервисы с уже готовыми интеграциями для:

• Sentry — отличный инструмент для траблшутинга

• Opentelemetry — собирает для нас трейсы

• Prometheus — собирает метрики

• и не только…

В основе Microbootstrap лежит паттерн settings, который позволяет декларативно задавать параметры для вышеперечисленных инструментов. Вот пример его использования с fastapi:

# settings.py
from microbootstrap import FastApiSettings

class YourSettings(FastApiSettings):
    ...  # Все ваши настройки здесь

settings = YourSettings()


# application.py
import fastapi
from microbootstrap.bootstrappers.litestar import FastApiBootstrapper

from your_application.settings import settings

# Готовое приложение с Sentry, Opentelemetry, Prometheus и т.д.
application: fastapi.FastAPI = FastApiBootstrapper(settings).bootstrap()

Пайплайны

У нас также есть централизованный пайплайн, который сильно облегчает создание нового сервиса. Ведь все, что надо сделать для работающего пайплайна — это добавить три строчки в .gitlab-ci.yml.

Вот так выглядит его подключение:

include:
  - project: "python-community/pypelines"
    file: "presets/backend--v1.yml"

Всего за три строчки мы получаем следующие стадии:

• статический анализ

• сборка

• тестирование

• сканирование на уязвимости

• релиз образа

• деплой

• запуск Е2Е тестов

У пайплайнов есть подробная документация ^[32]. Для более четкого понимания можно еще посмотреть доклад ^[33] о них.

Вывод

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

Наш подход к архитектуре призван помочь в разработке сервисов на Python. С его помощью мы закрываем у себя большинство базовых вопросов, которые могут возникнуть в процессе разработки. Конечно, у всех своя специфика. Поэтому мы постарались предложить практичные решения, которые подходят для большинства случаев. Плюс решения всегда можно адаптировать и улучшить в зависимости от специфики вашего проекта.

Мы надеемся, что эта статья поможет вам в непростом деле разработки сервисов и будет полезна в решении практических задач на каждом этапе разработки.