Привет, Хабр! Меня зовут Сергей, я техлид в команде PT BlackBox. Мы с коллегами разрабатываем продукт, который позволяет обнаруживать уязвимости в приложениях методом черного ящика. Фактически мы сами и пишем веб-приложения, и именно о них пойдет речь в статье.
Я бы хотел с вами поделиться своими наработками по теме бэкенд-приложений и предоставить вам шаблон-прототип, который, как мне кажется, может закрыть подавляющее большинство потребностей при их разработке.
Для написания прототипа я буду использовать Litestar, FastStream и dishka. Эта статья будет особенно полезна тем, кто пишет на Tornado, Django, Flask или AIOHTTP и хочет перейти на более актуальные технологии для дальнейшего развития своих проектов на современных рельсах.
Для быстрого перехода к разделам
Использование Litestar, FastStream и dishka позволяет успешно следовать принципам чистой архитектуры (сlean architecture) и почти не допускать «протекания» абстракций. В самом прототипе также будет использоваться Pydantic и SQLAlchemy, но я не буду уделять им особого внимания ввиду того, что эти технологии довольно популярны и в сообществе уже есть огромное количество гайдов по их применению.
Litestar
Litestar — это легковесный ASGI-фреймворк с подробной и проработанной документацией, на данный момент один из самых производительных ASGI-фреймворков.
Litestar — перспективный бэкенд-фреймворк с большим количеством «батареек» на разные случаи жизни. Фреймворк активно разрабатывается и поддерживается сообществом, но при всем этом обделен вниманием на Хабре, не считая анонса доклада на PHDays Fest 2.
FastStream
FastStream — это event source фреймворк, который упрощает и унифицирует работу с брокерами, а также избавляет от необходимости каждый раз изобретать велосипед. В нем есть валидация, автогенерация AsyncAPI, CLI-утилита поверх ChatGPT, которая генерирует готовые FastStream-сервисы по вашему описанию. Фреймворк поддерживает такие брокеры, как Kafka, RabbitMQ, NATS, Redis Stream.
Чистая архитектура (SOLID)
Как уже упоминалось выше, прототип построен на принципах чистой архитектуры. При его написании я руководствовался принципами SOLID и делением приложения на слои.
Для тех, кто не знаком с концепцией чистой архитектуры или хочет освежить ее в памяти, я подготовил этот раздел. Хочу заметить, что это не ультимативный гайд по чистой архитектуре и некоторые вещи могут быть намеренно упрощены. Вот иллюстрация чистой архитектуры, созданная Робертом Мартином:
Каждая окружность соответствует различным составляющим приложения. Чем ближе слой к центру, тем менее он подвержен изменениям, поэтому нижестоящие слои не должны зависеть от вышестоящих.
Frameworks and Drivers
В этом слое размещены внешние сущности, например конкретная база данных или интерфейс вашего приложения. В моем прототипе я не завязывал бизнес-логику приложения на этот слой.
Interface Adapters
Здесь размещены конкретные реализации, имплементирующие общение с внешними сервисами, например с базой данных. Можно представить это так:
class BooksGateway: def get_books(self) -> list[Books]: query = 'SELECT * FROM books' rows = self.psql.execute(query) return [Books(**r) for r in rows]Application business rules
В этом слое размещен код для реализации пользовательских сценариев. Здесь находятся сущности, которые описывают конкретный use case, агрегировав в себе необходимые зависимости. Схематично это можно изобразить так:
def interactor(self) -> bool: """ User buy books """ books = self.get_books_from_card() full_price = sum([b.price for b in books]) self.write_off_money(full_price) self.send_books_to_user(books)Enterprise business rules
Доменный слой приложения, в котором располагается основа приложения.
SOLID
Чистая архитектура предполагает не наличие слоев как таковых, а следование принципам SOLID. Слои — это всего лишь следствие, вытекающее из SOLID и парочки других принципов.
Это принципы построения любых приложений, они универсальны и не зависят от языка. Можно назвать их принципами вытекающими из ООП, хотя это не совсем корректно, потому что все эти принципы подходят для проектирования как микросервисной архитектуры, так и конкретного приложения. Исключительно к ООП можно отнести только один из пяти принципов.
S — Single responsibility principle
Принцип единой ответственности: модуль должен иметь одну и только одну причину для изменения.
O — Open-closed principle
Принцип открытости — закрытости: он означает, что должна быть возможность добавлять новые компоненты без изменения уже существующих.
L — Liskov substitution principle
Принцип подстановки Барбары Лисков заключается в том, что дочерний класс должен соответствовать поведению родителя. Это, собственно, тот принцип, который актуален только для ООП.
I — Interface segregation principle
Принцип разделения интерфейсов: компоненты не должны зависеть от методов, которые они не используют.
D — Dependency inversion principle
Принцип инверсии зависимостей: компоненты должны зависеть от абстрактных интерфейсов, а не от конкретных реализаций.
Dependency injection (dishka)
Dependency injection — это принцип, который позволяет уменьшить связанность (coupling) и увеличить сплоченность (cohesion) сущностей в приложении. Это происходит за счет следующего: если сущность А требует В для своей работы, она должна получить ее извне, а не создавать самостоятельно.
Для лучшего понимания можно провести следующую аналогию: связанность — это суперклей, которым мы соединили части нашего приложения, а сплоченность скорее похожа на конструктор. Подробнее об этом принципе можно узнать из статьи.
Пример кода с высокой связанностью:
import os class ApiClient: def __init__(self) -> None: self.api_key = os.getenv("API_KEY") # <-- dependency self.timeout = int(os.getenv("TIMEOUT")) # <-- dependency class Service: def __init__(self) -> None: self.api_client = ApiClient() # <-- dependency def main() -> None: service = Service() # <-- dependency ... if __name__ == "__main__": main()Тот же самый код, но с использованием принципа внедрения зависимостей:
import os class ApiClient: def __init__(self, api_key: str, timeout: int) -> None: self.api_key = api_key # <-- dependency is injected self.timeout = timeout # <-- dependency is injected class Service: def __init__(self, api_client: ApiClient) -> None: self.api_client = api_client # <-- dependency is injected def main(service: Service) -> None: # <-- dependency is injected ... if __name__ == "__main__": main( service=Service( api_client=ApiClient( api_key=os.getenv("API_KEY"), timeout=int(os.getenv("TIMEOUT")), ), ), )По ходу написания прототипа я придерживался принципа внедрения зависимостей.
Dishka — IoC-контейнер, фреймворк, который позволяет управлять зависимостями в проекте. Несмотря на то что и у Litestar, и у FastStream есть свои решения для этого, использование внешней технологии позволит запустить их в рамках одной кодовой базы.
Подробнее с принципом можно ознакомиться тут. В том же материале можно прочитать о том, что такое IoC-контейнер.
Реализация прототипа
В рамках этой статьи я написал прототип приложения, которое позволяет хранить информацию о книгах. Оно «слушает» очередь, в которую приходят события (events) и записывает информацию из этих событий в PostgreSQL.
Ниже я не буду подробно останавливаться на таких вещах, как написание docker compose или настройка миграций — на GitHub размещена готовая реализация, в которой все это сделано.
. ├── docker-compose.yaml — здесь поднимем RabbitMQ и Postgress ├── .env — здесь опишем переменное окружение └── book-club ├── domain — доменные сущности ├── application — здесь у нас будет лежать бизнес-логика приложения ├── infrastructure — здесь у нас будут хранится адаптеры и миграции ├── handlers — HTTP- и AMQP-контроллеры и DTO для них ├── ioc.py — контейнер с зависимостями ├── config.py — конфигурация проекта ├── main.py — точка входа в приложениеПервое, с чего хотелось бы начать, — описание домена. Домен является основой любого приложения — того, ради чего оно пишется. Например: приложение по доставке пиццы пишется ради пиццы, приложение автосервиса — ради машины.
Для лучшего понимания я специально сильно упрощаю. На самом деле домен приложения куда шире, он может включать в себя клиентов, которые покупают пиццу, сотрудников, которые обслуживают машину, побочные домены для продажи напитков и салонных елочек. Для книжного клуба я ограничился одной доменной сущностью — BookDM.
# book_club/domain/entities.py from dataclasses import dataclass @dataclass(slots=True) class BookDM: # DM - Domain model uuid: str title: str pages: int is_read: boolСледующим шагом я описал бизнес-логику приложения. Это то, чем приложение занимается. Приложение может сохранить информацию о книге и выдать ее по запросу. Рассмотрим DTO, который необходим для слоя бизнес-логики.
# book_club/application/dto.py from dataclasses import dataclass @dataclass(slots=True) class NewBookDTO: title: str pages: int is_read: boolDTO — это паттерн, который реализует обмен информацией между слоями. Внутренний слой не должен зависеть от внешнего, поэтому он не может работать с объектами из внешних слоев приложения, такими как pydantic model или объект request. NewBookDTO необходим, чтобы получать информацию о книгах из внешних слоев приложения.
Далее, исходя из бизнес-логики, я реализовал use-case interactor и описал интерфейсы, от которых они зависят.
# book_club/application/interfaces.py from abc import abstractmethod from typing import Protocol from uuid import UUID from book_club.domain.entities import BookDM class BookSaver(Protocol): @abstractmethod async def save(self, book: BookDM) -> None: ... class BookReader(Protocol): @abstractmethod async def read_by_uuid(self, uuid: str) -> BookDM | None: ... class UUIDGenerator(Protocol): def __call__(self) -> UUID: ... class DBSession(Protocol): @abstractmethod async def commit(self) -> None: ... @abstractmethod async def flush(self) -> None: ...
BookSaver— сохранение информации о книгах.
BookReader— чтение информации о книгах.
UUIDGenerator— генерация UUID.
DBSession— интерфейс для работы с транзакциями.
# book_club/application/interactors.py from book_club.application import interfaces from book_club.application.dto import NewBookDTO from book_club.domain import entities class GetBookInteractor: def __init__( self, book_gateway: interfaces.BookReader, ) -> None: self._book_gateway = book_gateway async def __call__(self, uuid: str) -> entities.BookDM | None: return await self._book_gateway.read_by_uuid(uuid) class NewBookInteractor: def __init__( self, db_session: interfaces.DBSession, book_gateway: interfaces.BookSaver, uuid_generator: interfaces.UUIDGenerator, ) -> None: self._db_session = db_session self._book_gateway = book_gateway self._uuid_generator = uuid_generator async def __call__(self, dto: NewBookDTO) -> str: uuid = str(self._uuid_generator()) book = entities.BookDM( uuid=uuid, title=dto.title, pages=dto.pages, is_read=dto.is_read ) await self._book_gateway.save(book) await self._db_session.commit() return uuidUse-case interactor — это реализация конкретного пользовательского действия. У пользователя только два доступных действия, которые и описаны в теле интеракторов.
На этом этапе основа приложения уже готова, теперь рассмотрим детали реализации.
Config приложения:
# book_club/config.py from os import environ as env from pydantic import Field, BaseModel class RabbitMQConfig(BaseModel): host: str = Field(alias='RABBITMQ_HOST') port: int = Field(alias='RABBITMQ_PORT') login: str = Field(alias='RABBITMQ_USER') password: str = Field(alias='RABBITMQ_PASS') class PostgresConfig(BaseModel): host: str = Field(alias='POSTGRES_HOST') port: int = Field(alias='POSTGRES_PORT') login: str = Field(alias='POSTGRES_USER') password: str = Field(alias='POSTGRES_PASSWORD') database: str = Field(alias='POSTGRES_DB') class Config(BaseModel): rabbitmq: RabbitMQConfig = Field(default_factory=lambda: RabbitMQConfig(**env)) postgres: PostgresConfig = Field(default_factory=lambda: PostgresConfig(**env)) Обращаю внимание, что я разделил config на отдельные составляющие, а потом объединил их путем композиции в классе Config. Такой подход позволяет более эффективно проводить инъекцию зависимостей. Если класс зависит от конкретных конфигурационных данных, то нужно предоставить ему только эти данные вместо конфигурации всего приложения.
# book_club/infrastructure/database.py from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine from book_club.config import PostgresConfig def new_session_maker(psql_config: PostgresConfig) -> async_sessionmaker[AsyncSession]: database_uri = 'postgresql+psycopg://{login}:{password}@{host}:{port}/{database}'.format( login=psql_config.login, password=psql_config.password, host=psql_config.host, port=psql_config.port, database=psql_config.database, ) engine = create_async_engine( database_uri, pool_size=15, max_overflow=15, connect_args={ "connect_timeout": 5, }, ) return async_sessionmaker(engine, class_=AsyncSession, autoflush=False, expire_on_commit=False)Обратите внимание, new_session_maker зависит только от PostgresConfig, он ничего не знает о RabbitMQConfig.
# book_club/infrastructure/broker.py from faststream.rabbit import RabbitBroker from faststream.security import SASLPlaintext from book_club.config import RabbitMQConfig def new_broker(rabbitmq_config: RabbitMQConfig) -> RabbitBroker: return RabbitBroker( host=rabbitmq_config.host, port=rabbitmq_config.port, security=SASLPlaintext( username=rabbitmq_config.login, password=rabbitmq_config.password, ), virtualhost="/", )Аналогичная ситуация и для new_broker.
# book_club/infrastructure/gateways.py from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import text from book_club.application.interfaces import BookReader, BookSaver from book_club.domain.entities import BookDM class BookGateway( BookReader, BookSaver, ): def __init__(self, session: AsyncSession): self._session = session async def read_by_uuid(self, uuid: str) -> BookDM | None: query = text("SELECT * FROM books WHERE uuid = :uuid") result = await self._session.execute( statement=query, params={"uuid": uuid}, ) row = result.fetchone() if not row: return None return BookDM( uuid=row.uuid, title=row.title, pages=row.pages, is_read=row.is_read, ) async def save(self, book: BookDM) -> None: query = text("INSERT INTO books (uuid, title, pages, is_read) VALUES (:uuid, :title, :pages, :is_read)") await self._session.execute( statement=query, params={ "uuid": book.uuid, "title": book.title, "pages": book.pages, "is_read": book.is_read, }, )Gateway инкапсулирует в себе работу с данными и позволяет работать с ними как с коллекцией, не вдаваясь в детали реализации.
Отмечу, что BookGateway удовлетворяет упомянутым ранее интерфейсам BookSaver и BookReader.
После реализации домена и адаптеров, я поместил зависимости в контейнер для их последующей инъекции.
# book_club/ioc.py from typing import AsyncIterable from uuid import uuid4 from dishka import Provider, Scope, provide, AnyOf, from_context from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from book_club.application import interfaces from book_club.application.interactors import ( GetBookInteractor, NewBookInteractor ) from book_club.config import Config from book_club.infrastructure.database import new_session_maker from dbook_club.infrastructure.gateways import BookGateway class AppProvider(Provider): ...Таким образом, я описал провайдер для interfaces.UUIDGenerator, у него есть параметр scope. Этот параметр определяет срок жизни зависимости (его подробное описание можно найти в документации). Зависимость, у которой область видимости APP, не нуждается в пересоздании. Здесь я использую функцию uuid4 как объект первого типа, он удовлетворяет ранее описанному interfaces.UUIDGenerator. В случае с uuid4 достаточно создать объект один раз за все время жизни приложения, поэтому у него scope=Scope.APP.
... @provide(scope=Scope.APP) def get_uuid_generator(self) -> interfaces.UUIDGenerator: return uuid4 ...Далее я описал provider для config, у него тоже выставлен параметр scope=Scope.APP. Как и в случае с uuid4, нам достаточно создать объект один раз за все время жизни приложения. Однако from_context означает, что объект Config должен быть создан где-то вовне и передан при создании AppProvider.
... config = from_context(provides=Config, scope=Scope.APP) ...Следующий шаг — это создание sessionmaker из SqlAlchemy.
... @provide(scope=Scope.APP) def get_session_maker(self, config: Config) -> async_sessionmaker[AsyncSession]: return new_session_maker(config.postgres) ...Тут я тоже выставил scope=Scope.APP, поскольку класс async_sessionmaker нам больше пересоздавать не требуется. Этот провайдер зависит от Config, что я и указал в сигнатуре функции; async_sessionmaker нужен для создания AsyncSession.
... @provide(scope=Scope.REQUEST) async def get_session(self, session_maker: async_sessionmaker[AsyncSession]) -> AsyncIterable[AnyOf[ AsyncSession, interfaces.DBSession, ]]: async with session_maker() as session: yield session ...Здесь уже scope=Scope.REQUEST. Этот объект будет создаваться каждый раз при вызове, по сути, после на каждый запрос пользователя будет создаваться новая сессия алхимии. Сигнатура этого провайдера говорит о том, что он зависит от async_sessionmaker[AsyncSession] и возвращаемый объект должен удовлетворять сигнатуре сразу двух интерфейсов: sqlalchemy.AsyncSession и interfaces.DBSession.
Использование генераторов позволяет хранить состояние сессии, а после ответа на запрос пользователя — вернуться в этот провайдер и выйти из контекстного менеджера, тем самым вернуть ресуры в connection pool.
... book_gateway = provide( BookGateway, scope=Scope.REQUEST, provides=AnyOf[interfaces.BookReader, interfaces.BookSaver] ) ...Этот провайдер отвечает за создание BookGateway, который зависит от sqlalchemy.AsyncSession и должен удовлетворять интерфейсам interfaces.BookReader и interfaces.BookSaver. От этих интерфейсов зависят interactors, создание которых выглядит так:
... get_book_interactor = provide(GetBookInteractor, scope=Scope.REQUEST) create_new_book_interactor = provide(NewBookInteractor, scope=Scope.REQUEST) ...Под капотом dishka вызывает конструкторы для BookGateway, GetBookInteractor и NewBookInteractor, сигнатура которых зависит от interfaces.BookReader, interfaces.BookSaver, interfaces.DBSession, interfaces.UUIDGenerator. В случае если у зависимостей выставлен параметр Scope.REQUEST, они будут созданы, если же выставлен Scope.APP, то dishka попытается найти их в своем хранилище. Если в нем нет зависимостей — dishka создаст их и разместит там.
При получении какого-либо события (будь то запрос по HTTP или сообщение из очереди) вызывается обработчик, который запускает цепочку создания зависимостей, а после окончания события происходит их финализация. В такой схеме на каждое событие будет только одно подключение к базе данных для всех зависимостей в этой цепочке, что позволит достигнуть атомарности транзакций и эффективного использования ресурсов пула подключений.
Теперь обратим внимание на реализацию слоя контроллеров.
BookSchema — это DTO, который мы используем для общения между слоем контроллеров и внешним слоем. Несмотря на то что в прототипе два контроллера, нам нужна всего одна модель, которая одинаково хорошо поддерживается как FastStream, так и Litestar.
На основе только этой модели строится валидация и автогенерация документации.
# controllers/schemas.py from pydantic import BaseModel class BookSchema(BaseModel): title: str pages: int is_read: bool# book_club/controllers/amqp.py from dishka.integrations.base import FromDishka as Depends from faststream.rabbit import RabbitRouter from book_club.application.dto import NewBookDTO from book_club.application.interactors import NewBookInteractor from book_club.controllers.schemas import BookSchema AMQPBookController = RabbitRouter() @AMQPBookController.subscriber("create_book") @AMQPBookController.publisher("book_statuses") async def handle(data: BookSchema, interactor: Depends[NewBookInteractor]) -> str: dto = NewBookDTO( title=data.title, pages=data.pages, is_read=data.is_read ) uuid = await interactor(dto) return uuidВыше приведена реализация контроллера для сохранения информации о книге. Как видно из кода, контроллер получает на вход BookSchema, контроллер «слушает» очередь create_book и, в случае наличия события в очереди, пытается превратить данные оттуда в BookSchema. Если не получается, то вызывается исключение pydantic.ValidationError.
В случае успеха происходит вызов handle, который зависит от NewBookInteractor, и создание цепочки зависимостей, о которой я говорил выше. После этого отрабатывает интерактор и в очередь book_statuses отправляется UUID созданной записи о книге.
# book_club/controllers/http.py from typing import Annotated from uuid import UUID from dishka.integrations.base import FromDishka as Depends from dishka.integrations.litestar import inject from litestar import Controller, route, HttpMethod from litestar import status_codes from litestar.exceptions import HTTPException from litestar.params import Body from book_club.application.interactors import GetBookInteractor from book_club.controllers.schemas import BookSchema class HTTPBookController(Controller): path = "/book" @route(http_method=HttpMethod.GET, path="/{book_id:uuid}") @inject async def get_book( self, book_id: Annotated[UUID, Body(description="Book ID", title="Book ID")], interactor: Depends[GetBookInteractor], ) -> BookSchema: book_dm = await interactor(uuid=str(book_id)) if not book_dm: raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Book not found") return BookSchema( title=book_dm.title, pages=book_dm.pages, is_read=book_dm.is_read, )Выше представлена реализация контроллера, который отвечает за выдачу информации о книгах. Логика построения цепочки зависимостей абсолютно такая же, как и в примере выше. На вход ожидается book_id в виде UUID, по которому идет поиск книги в базе данных.
Теперь можно все собрать, и точка входа в приложение будет выглядеть следующим образом:
# book_club/main.py from dishka import make_async_container from dishka.integrations import faststream as faststream_integration from dishka.integrations import litestar as litestar_integration from faststream import FastStream from litestar import Litestar from book_club.config import Config from book_club.controllers.amqp import AMQPBookController from book_club.controllers.http import HTTPBookController from book_club.infrastructure.broker import new_broker from book_club.ioc import AppProvider config = Config() container = make_async_container(AppProvider(), context={Config: config}) def get_faststream_app() -> FastStream: broker = new_broker(config.rabbitmq) faststream_app = FastStream(broker) faststream_integration.setup_dishka(container, faststream_app, auto_inject=True) broker.include_router(AMQPBookController) return faststream_app def get_litestar_app() -> Litestar: litestar_app = Litestar( route_handlers=[HTTPBookController], ) litestar_integration.setup_dishka(container, litestar_app) return litestar_app def get_app(): faststream_app = get_faststream_app() litestar_app = get_litestar_app() litestar_app.on_startup.append(faststream_app.broker.start) litestar_app.on_shutdown.append(faststream_app.broker.close) return litestar_appConfig и container вынесены в глобальную область видимости, это необходимо, чтобы была возможность запустить HTTP и AMQP-часть изолированно друг от друга в разных процессах, если вдруг возникнет такая необходимость.
Теперь это приложение полностью готово, и его можно запустить и проверить.
Обратите внимание, что для корректной работы приложения необходимы контейнеры c RabbitMQ и Postgress, виртуальное окружение и миграции. Подробные инструкции, docker-compose.yaml и миграции вы найдете в исходном коде проекта.
Запускаем приложение:
uvicorn --factory book_club.main:get_app --reloadПишем в очередь create_book сообщение:
rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS \ publish exchange=amq.default routing_key=create_book \ payload='{"title": "The Brothers Karamazov", "pages": 928, "is_read": true}'Читаем из очереди book_statuses сообщение и получаем UUID:
rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS get queue=book_statuses count=1 +---------------+--------------------------------------+ | routing_key | payload | +---------------+--------------------------------------+ | book_statuses | bc5fe42d-bedd-42e4-b518-89d088808237 | +---------------+--------------------------------------+С этим UUID идем на HTTP API и забираем информацию о книге:
curl http://localhost:8000/book/bc5fe42d-bedd-42e4-b518-89d088808237 {"title":"The Brothers Karamazov","pages":928,"is_read":true}Если вы хотите запустить HTTP и AMQP в разных процессах, это можно сделать вот так:
faststream run --factory book_club.main:get_faststream_app --reload uvicorn --factory book_club.main:get_litestar_app --reloadЗаключение
Спасибо за внимание! Призываю вас зайти в исходный код прототипа и изучить его самостоятельно. Если этот туториал окажется полезным (я это пойму по плюсикам к статье), выпущу вторую часть с демонстрацией, как описанный проект тестировать. Также вступайте в комьюнити — там вы можете встретить контрибьюторов технологий, о которых шла речь в статье, и задать им интересующие вас вопросы: FastAPI / Litestar / BlackSheep (ASGI Community), FastStream Community, Reagento Community.
ссылка на оригинал статьи https://habr.com/ru/articles/820171/
Добавить комментарий