eventbus-lite 1.0.0

Lightweight thread-safe synchronous event bus for Python

EventBus

Лёгкая потокобезопасная шина событий (publish/subscribe) для Python.

Возможности

• синхронная обработка событий;
• потокобезопасность;
• приоритеты обработчиков;
• стабильный порядок вызова;
• одноразовые подписки (once);
• декораторы регистрации;
• временные подписки через контекстный менеджер;
• поддержка шаблонов * и **;
• отмена события;
• остановка распространения события;
• слабые ссылки на методы и вызываемые объекты;
• автоматическое удаление уничтоженных подписчиков;
• кэширование маршрутизации событий;
• генерация событий об ошибках;
• строгий режим обработки исключений;
• отсутствие внешних зависимостей.

Поддерживаемые версии Python:

• Python 3.10+
• Python 3.11
• Python 3.12
• Python 3.13

---

Установка

pip install eventbus

---

Быстрый старт

from eventbus import EventBus

bus = EventBus()


def handler(event):
    print(event.type)
    print(event.data)


bus.subscribe(
    "system/start",
    handler
)

bus.publish(
    "system/start",
    version="1.0"
)

Результат:

system/start
{'version': '1.0'}

---

Объект Event

Каждое событие представлено экземпляром класса Event.

event.type
event.data
event.source
event.timestamp
event.result
event.cancelled
event.propagation_stopped
event.exception

---

Подписка

bus.subscribe(
    "system/start",
    handler
)

---

Публикация

event = bus.publish(
    "system/start",
    value=123
)

Метод publish() возвращает объект Event.

---

Приоритеты

Обработчики вызываются в порядке возрастания приоритета.

bus.subscribe(
    "test",
    first_handler,
    priority=100
)

bus.subscribe(
    "test",
    second_handler,
    priority=200
)

Сначала будет вызван first_handler.

---

Стабильный порядок

Если приоритеты одинаковы, обработчики вызываются в порядке регистрации.

bus.subscribe("test", h1)
bus.subscribe("test", h2)
bus.subscribe("test", h3)

Порядок вызова:

h1
h2
h3

---

Одноразовые обработчики

bus.subscribe(
    "system/start",
    handler,
    once=True
)

После первого вызова обработчик будет автоматически удалён.

---

Декораторы

on()

@bus.on("system/start")
def on_start(event):
    print("started")

once()

@bus.once("system/start")
def initialize(event):
    print("called once")

---

Удаление подписки

bus.unsubscribe(
    "system/start",
    handler
)

---

Контекстный менеджер

with bus.subscription(
    "system/start",
    handler
):
    bus.publish("system/start")

# подписка автоматически удалена

---

Шаблоны

Точное совпадение

system/start

---

*

Один сегмент.

system/*

Подходит для:

system/start
system/stop

---

**

Любое количество сегментов.

system/**

Подходит для:

system
system/start
system/core/start
system/core/network/start

---

Отмена события

def handler(event):
    event.cancel("denied")

После вызова:

event.cancelled == True

Оставшиеся обработчики вызваны не будут.

Результат доступен через:

event.result

---

Остановка распространения

def handler(event):
    event.stop_propagation()

Оставшиеся обработчики не вызываются, однако событие не считается отменённым.

event.cancelled == False

---

Исключения

По умолчанию исключения внутри обработчиков не прерывают работу EventBus.

def handler(event):
    raise RuntimeError()

Исключение сохраняется:

event.exception

Также автоматически публикуется событие:

system/error/event

---

Строгий режим

bus = EventBus(strict=True)

В этом режиме исключение приводит к возбуждению:

EventDispatchError

---

Слабые ссылки

Методы объектов и вызываемые экземпляры хранятся через слабые ссылки.

Если объект уничтожен сборщиком мусора, подписка удаляется автоматически.

class Receiver:

    def on_event(self, event):
        pass


receiver = Receiver()

bus.subscribe(
    "test",
    receiver.on_event
)

---

Потокобезопасность

Все операции регистрации, удаления подписчиков и публикации событий являются потокобезопасными.

from threading import Thread

Один экземпляр EventBus может безопасно использоваться несколькими потоками одновременно.

---

Исключения библиотеки

EventBusError

Базовый класс всех исключений.

InvalidPatternError

Некорректный шаблон события.

InvalidHandlerError

Переданный объект не является вызываемым.

DuplicateSubscriptionError

Ошибка повторной регистрации.

EventDispatchError

Ошибка обработки события.

---

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

Проект покрыт тестами.

pytest

---

Лицензия

MIT License