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

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

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

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

Код стайл

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

• 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, про нее можно прочитать здесь

API

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

Организация 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 можно почитать здесь

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

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

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

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

Альтернативно можно версионировать с помощью пути — /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, но он долгое время не поддерживал Python 3.12 и мы перебрались на that-depends. Этот пакет предлагает все те же концепции, что и dependency-injector, но при этом намного меньше и проще.

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

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

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

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

За последний год большую популярность получила библиотека FastStream. Она значительно упрощает создание консьюмеров и хорошо подходит нам, так как мы в основном используем Kafka и NATS. FastStream автоматически генерирует AsyncAPI документацию (AsyncAPI) для наших консьюмеров. Это аналог Swagger для 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 мы тратим меньше сил на создание консьюмеров и можем больше сосредоточиться на логике приложения. Если ещё не пользовались, почитайте официальную документацию.

Конечно, учитывая, что мы работаем в банке, у нас есть много легаси-инфраструктуры, например, ActiveMQ Artemis. Для таких случаев мы разработали собственный пакет, совместимый с 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 + Alembic как самую популярную и проверенную временем.

Наш лейаут:

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. Она предоставляет базовый класс со всеми необходимыми 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.

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

import stamina

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

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

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

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

from circuitbreaker import circuit

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

Тесты

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

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

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

• 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, а также собираем отчет о тестировании при помощи pytest-cov.

Microbootstrap

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

С помощью 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Е тестов

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

Вывод

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

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

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