src/liquidcode/event_dispatcher.py - develop-0.4 - liquid-g/liquid-code

src/liquidcode/event_dispatcher.py

70 строк2 KB

User

0.4.5 - feat: добавить Event Dispatcher и API слушателей для жизненного цикла запроса

05 июл 2026, 14:53

05 июл 2026, 14:5367c6f20

"""
Простой Event Dispatcher для LiquidCode.
"""

from typing import Any, Callable, Dict, List


class Event:
    """Базовый класс для событий."""
    def __init__(self, name: str, data: Any = None):
        self.name = name
        self.data = data
        self.propagation_stopped = False

    def stop_propagation(self) -> None:
        """Останавливает дальнейшее распространение события."""
        self.propagation_stopped = True


class EventDispatcher:
    """
    Диспетчер событий. Позволяет регистрировать слушатели и отправлять события.
    """

    def __init__(self):
        self._listeners: Dict[str, List[Callable]] = {}

    def add_listener(self, event_name: str, listener: Callable, priority: int = 0) -> None:
        """
        Добавляет слушатель для события.

        Args:
            event_name: Имя события.
            listener: Callable, принимающий событие.
            priority: Чем выше число, тем раньше вызывается слушатель.
        """
        if event_name not in self._listeners:
            self._listeners[event_name] = []
        self._listeners[event_name].append((priority, listener))
        # Сортируем по убыванию приоритета
        self._listeners[event_name].sort(key=lambda x: x[0], reverse=True)

    def remove_listener(self, event_name: str, listener: Callable) -> None:
        """Удаляет слушатель для события."""
        if event_name in self._listeners:
            self._listeners[event_name] = [
                (p, l) for p, l in self._listeners[event_name] if l != listener
            ]
            if not self._listeners[event_name]:
                del self._listeners[event_name]

    def dispatch(self, event: Event) -> Event:
        """
        Отправляет событие всем зарегистрированным слушателям.

        Args:
            event: Объект события.

        Returns:
            Объект события (возможно, изменённый слушателями).
        """
        if event.name not in self._listeners:
            return event

        for _, listener in self._listeners[event.name]:
            listener(event)
            if event.propagation_stopped:
                break

        return event