Skip to main content

Набор компонентов для построения многослойной архитектуры

Project description

Explicit

Набор компонентов для построения явной (Explicit) многослойной архитектуры

explicit architecture

Решаемые проблемы

Основные принципы построения

  • CQRS:
    • Поток команд (Command), модифицирующий состояние приложения (БД) идёт через слой предметной области
    • Поток запросов (Query) не меняет состояние и не проходит через предметную область
  • В основе лежит Явная архитектура: слой предметной области не зависит от сервисного слоя или конкретных реализаций ORM, СУБД, API сторонних служб и т.д.; используются порты и адаптеры абстрагируют вызовы к API; ядро приложения отделено от взаимодействующих с ним API, GUI, webservices.

Реализуемые компоненты

Команда (command)

Инкапсулирует передачу параметров запроса обработчику команды, стандартизирует передачу параметров:

  # Application core
  class RegisterStudent(Command):
    last_name: str
    first_name: str

Обработчик команды (command handler)

Принимает команду и выполняет действия над переданными в команде данными.

  # Application core
  def register_student(command: RegisterStudent):  # handler
      student = Student(last_name=command.last_name, first_name=command.first_name)
      repository.add(student)

Шина (bus, messagebus)

Обеспечивает доставку команды соответствующему обработчику, уменьшает зацепление между модулями и слоями приложения.

  from core import bus

  # Django Rest Framework
  class StudentViewSet(ModelViewSet):
  
    def perform_create(self, serializer):
        command = RegisterStudent(**serializer.validated_data)
        bus.handle(command)
    
  # Spyne webservices
  @rpc(
    StudentData,
  )
  def RegisterStudent(ctx, data: 'StudentData'):
    command = RegisterStudent(last_name=data.last_name, first_name=data.first_name)
    bus.handle(command)
  

Unit of Work

Unit of Work:

  • Единица работы, логическая бизнес-транзация
  • Обеспечивает атомарность выполняемых операций
  • Предоставляет доступ к репозиториям приложения
  • Устраняет зависимость логики от конкретного фреймворка или технологии БД
def register_student(command: RegisterStudent, uow: 'UnitOfWork'):
    with uow.wrap():
        uow.users.add(user)
        uow.persons.add(person)
        uow.students.add(student)

Репозиторий (Repository)

Repository:

  • Является адаптером к СУБД
  • Выполняет роль хранилища объектов предметной области в соответствующем слое
  • Инкапсулирует логику выборки данных
  • Устраняет зависимость логики от конкретного фреймворка или технологии БД
class Repository:
    def get_object_by_id(self, identifier: int) -> Student:
        try:
            dbinstance = DBStudent.objects.get(pk=identifier)
            return self._to_domain(dbinstance)
        except ObjectDoesNotExist as e:
            raise StudentNotFound() from e

    def get_by_persons(self, *persons: 'Person') -> Generator[Student, None, None]:
        query = DBStudent.objects.filter(person_id__in=(person.id for person in persons))
        for dbinstance in query.iterator():
            yield self._to_domain(dbinstance)

Фабрика (Factory)

Инкапсулирует логику создания нового объекта предметной области по известным параметрам.

class Factory(AbstractDomainFactory):
    def create(self, data: StudentDTO) -> Student:
        return Student(
            person_id=data.person.id,
            unit_id=data.unit.id,
        )

Объект передачи данных (DTO, Data Transfer Object)

DTO используется при передаче большого количества параметров между объектами приложения и для стандартизации передачи данных и контрактов.

def create_person(data: 'PersonDTO'):
    person = factory.create(data)

def update_person(person, data: 'PersonDTO'):
    domain.update_person(data)

Обработка входящего запроса и события

1. Обработка запроса

1.1 Запрос приходит в контроллер (View, Spyne @rpc и т.д.)

1.2 Контроллер извлекает параметры запроса, валидирует их, формирует команду с параметрами запроса

1.3 Сформированная команда направляется в шину ядра приложения

def post(self, request, *args, **kwargs):
  serializer = self.get_serializer(data=request.data)
  serializer.is_valid(raise_exception=True)
  last_name = serializer.validated_data['last_name']
  first_name = serializer.validated_data['first_name']

  command = RegisterStudent(
      last_name=last_name,
      first_name=first_name,
  )
  bus.handle(command)
  return JsonResponse(data={'registered': True})

1.4 Шина по типу команды определяет соответствующий обработчик и передает ему команду

class CommandBusMixin(ABC):

   _command_handlers: Dict[Type[Command], Callable] = {}
   
   def handle_command(self, command: Command):
      """Передача запроса обработчику типа данного запроса."""
      return self._command_handlers[type(command)](command)

1.5 Обработчик выполняет требуемые действия, инициирует событие, соответствующее результату обработки, возвращает результат

def register_student(
   command: RegisterStudent, uow: 'UnitOfWork'
) -> 'Student':
   with uow.wrap():
       student: Student = domain_register_student(
           StudentDTO(**command.dict())
       )
   uow.add_event(events.StudentCreated(
       **command.dict(), **asdict(student)
   ))
   return student

1.6 Событие попадает в шину ядра

1.7 Ядро определяет список обработчиков, соответствующих типу события, передает им инстанс события

class EventBusMixin(ABC):

   _event_handlers: Dict[Type[Event], List[EventHandler]] = {}

   def handle_event(self, event: Event):
       """Передача события списку обработчиков типа данного события."""
       consume(
           handler(event)
           for handler in self._event_handlers[type(event)]
       )

1.8 Обработчик события выполняет требуемые действия

1.9 Обработчик события может передать событие на внешнюю шину, воспользовавшись соответствующим адаптером шины

def on_student_created(
   event: events.StudentCreated, adapter: 'AbstractAdapter'
):
   adapter.publish(
       'edu.students.created',
       json.dumps(asdict(event), default=encoder)
   )

2. Обработка внешнего события

2.1 Подписчик получает событие из внешней шины

2.2 Подписчик десериализует внешнее событие и инстанцирует внутреннее

def bootstrap():
   from students.core.adapters.messaging import adapter
   from students.core.domain import events
   
   TOPIC_EVENTS = {
      'edu.persons.created': events.PersonCreated,
   }
   
   for message in adapter.subscribe(*TOPIC_EVENTS):
      event = TOPIC_EVENTS[message.topic()](
          **json.loads(message.value())
      )
      bus.handle(event)

2.3 Внутренее событие попадает в шину ядра

2.4 Ядро определяет список обработчиков, соответствующих типу события, передает им инстанс события

2.5 Обработчик события выполняет требуемые действия

2.6 Обработчик события может инстанцировать новое событие

def on_person_created(
   event: events.PersonCreated, uow: 'UnitOfWork'
) -> None:
   with uow.wrap():
       transaction_id = event.meta.transaction_id

       saga = uow.sagas.get_object_by_uuid(transaction_id)
       student = uow.students.get_object_by_id(saga.student_id)

       student.person_id = event.id
       uow.students.update(student)

       uow.add_event(events.StudentCreated(
           **command.dict(), **asdict(student)
       ))

Минимальный пример реализации

Можно увидеть в тестовом приложении src/testapp.

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

$ tox

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

explicit_python-2.2.1.tar.gz (20.3 kB view details)

Uploaded Source

Built Distribution

explicit_python-2.2.1-py3-none-any.whl (23.5 kB view details)

Uploaded Python 3

File details

Details for the file explicit_python-2.2.1.tar.gz.

File metadata

  • Download URL: explicit_python-2.2.1.tar.gz
  • Upload date:
  • Size: 20.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.8.0 pkginfo/1.10.0 readme-renderer/44.0 requests/2.32.3 requests-toolbelt/1.0.0 urllib3/2.2.3 tqdm/4.66.5 importlib-metadata/8.5.0 keyring/25.4.1 rfc3986/2.0.0 colorama/0.4.6 CPython/3.9.7

File hashes

Hashes for explicit_python-2.2.1.tar.gz
Algorithm Hash digest
SHA256 bf350308c98ca478aedbe2d28ae9a76cc21d2597d456aa6a40bb2fc18e6504c5
MD5 078328eed04c41383ad36194b786046b
BLAKE2b-256 744a6b43ff6d9252f97fa7df22c848bee298ecb1049b381645f5b7a56de5a9a9

See more details on using hashes here.

File details

Details for the file explicit_python-2.2.1-py3-none-any.whl.

File metadata

  • Download URL: explicit_python-2.2.1-py3-none-any.whl
  • Upload date:
  • Size: 23.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.8.0 pkginfo/1.10.0 readme-renderer/44.0 requests/2.32.3 requests-toolbelt/1.0.0 urllib3/2.2.3 tqdm/4.66.5 importlib-metadata/8.5.0 keyring/25.4.1 rfc3986/2.0.0 colorama/0.4.6 CPython/3.9.7

File hashes

Hashes for explicit_python-2.2.1-py3-none-any.whl
Algorithm Hash digest
SHA256 8d5563a7f6df5f4a148a7245192278aa1cfdda9aa1c930aa67a38acad60fb3d3
MD5 e89e35406e3e85b1dad862359fccd2c2
BLAKE2b-256 3d4bfe07b89f5f5bdf9927ba5e028b2b5767e680135e7f6013ddf9ec3a33b25d

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page