50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
# app/src/core/event_bus.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
class EventBus:
|
|
_version: int = 0
|
|
_last_event_type: str | None = None
|
|
_last_payload: dict[str, Any] = {}
|
|
_events: list[tuple[int, str, dict[str, Any]]] = []
|
|
_max_events: int = 100
|
|
|
|
# зафиксировать важное событие системы
|
|
@classmethod
|
|
def emit(cls, event_type: str, payload: dict[str, Any] | None = None) -> None:
|
|
cls._version += 1
|
|
cls._last_event_type = event_type
|
|
cls._last_payload = payload or {}
|
|
|
|
cls._events.append(
|
|
(
|
|
cls._version,
|
|
event_type,
|
|
dict(cls._last_payload),
|
|
)
|
|
)
|
|
|
|
if len(cls._events) > cls._max_events:
|
|
cls._events = cls._events[-cls._max_events:]
|
|
|
|
# текущая версия событий
|
|
@classmethod
|
|
def version(cls) -> int:
|
|
return cls._version
|
|
|
|
# последнее событие
|
|
@classmethod
|
|
def last_event(cls) -> tuple[str | None, dict[str, Any]]:
|
|
return cls._last_event_type, dict(cls._last_payload)
|
|
|
|
# события после указанной версии
|
|
@classmethod
|
|
def events_after(cls, version: int) -> list[tuple[int, str, dict[str, Any]]]:
|
|
return [
|
|
(event_version, event_type, dict(payload))
|
|
for event_version, event_type, payload in cls._events
|
|
if event_version > version
|
|
] |