Explicit-Kafka

Набор компонентов для интеграции explicit с kafka.

Содержит реализацию адаптера обмена сообщениями через Kafka.

Пример использования

Настройка адаптера

# persons/core/apps.py
from django.apps.config import AppConfig as AppConfigBase


class AppConfig(AppConfigBase):

    name = __package__
    
    def _setup_adapter(self):
        from explicit.kafka.adapters.messaging import Adapter
        from explicit.kafka.adapters.messaging import PublishConfig
        from explicit.kafka.adapters.messaging import SubscribeConfig

        from persons import core
        
        # конфигурация адаптера
        adapter_base_config = {'bootstrap.servers': 'kafka:9092'}
        publish_config = PublishConfig(adapter_base_config)
        subscribe_config = SubscribeConfig(adapter_base_config | {'group.id': f'edu.persons'})
    
        adapter = Adapter(subscribe_config=subscribe_config, publish_config=publish_config)
        core.adapter = adapter

    def ready(self):
        self._setup_adapter()

Отправка сообщений

# persons/core/persons/services/handlers/events.py

def on_person_created(
    event: 'PersonCreated',
    messaging_adapter: 'AbstractMessagingAdapter'
):
    messaging_adapter.publish('edu.persons.person', event.dump())

Подписка на сообщения

# education/entrypoints/eventconsumer.py

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

bootstrap()

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

Тесты запускаются через tox. В системе должен быть установлен uv.

Для корректного запуска тестов необходимы следующие значения переменных окружения в системе. Вместо <login> и <password> необходимо подставить данные от УЗ БАРС:

UV_INDEX="internal=http://<login>:<password>@nexus.py.bars.group/repository/pypi-all/simple"
UV_DEFAULT_INDEX="https://pypi.org/simple"
UV_INDEX_STRATEGY=unsafe-best-match
# Для docker 29+
TOX_DOCKER_GATEWAY=localhost

Чтобы не менять значения системных переменных, можно создать .env файл c переменными окружения и запускать команды через python-dotenv.

Установка tox с необходимыми зависимостями:

uv tool install tox --with tox-uv --with m3-tox-docker

Или с использованием python-dotenv:

dotenv run uv tool install tox --with tox-uv --with m3-tox-docker

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

tox

Через python-dotenv:

dotenv run tox