import builtins from collections import defaultdict from loguru import logger log = logger.bind(module="EventsSystem", prefix="init") log_reg = log.bind(prefix="register") log_unreg = log.bind(prefix="unregister") log_call = log.bind(prefix="call") class EventsSystem: def __init__(self): self.__events = defaultdict(list) log.debug("Injected to builtins") builtins.event = self log.debug("EventsSystem ready") def register(self, event_name, callback): """ Registers a callback for a given event name. :param event_name: Name of the event. :param callback: Callable to be executed when the event occurs. """ log_reg.debug(f"{event_name!r}; {callback.__name__}({callback})") if not callable(callback): return log_reg.error(f"Bad event '{event_name}' - cannot registered") self.__events[event_name].append(callback) def unregister(self, event_name, callback): """ Unregisters a callback for a given event name. :param event_name: Name of the event. :param callback: Callable to be unregistered. """ log_unreg.debug(f"{event_name!r}; {callback.__name__}({callback})") if event_name in self.__events.keys(): self.__events[event_name].remove(callback) else: log_unreg.error(f"Event '{event_name}' not found.") def call(self, event_name, *args, **kwargs): """ Call all callbacks for a given event name. :param event_name: Name of the event. :param args: Data to be passed to callbacks. :param kwargs: Additional data to be passed to callbacks. :return: List of data returned by callbacks. """ kwargs['event_name'] = event_name callbacks_data = [] if event_name in self.__events.keys(): for callback in self.__events[event_name]: log_call.debug(f"{event_name!r}; execute: '{callback.__name__}'; data: {args=}; {kwargs=}") try: event_data = callback(*args, **kwargs) callbacks_data.append(event_data) except Exception as e: log_call.error(f"Error in {callback.__name__}. See debug.log") log_call.exception(e) else: log_call.error(f"Event '{event_name}' not found.") return callbacks_data