HivenEventHandler and Listeners

---

Important

DispatchEventListener is the base Class for SingleDispatchEventListener and MultiDispatchEventListener, which should not be used

openhivenpy.events.DispatchEventListener

Base Class for all DispatchEventListeners

awaitable: Union[Awaitable, Callable] property readonly

client: HivenClient property readonly

event_name: str property readonly

Methods

__call__(self, *args, **kwargs) special

Returns the dispatch function of the class itself

Parameters:

Name	Type	Description	Default
event_data		Data of the received event	required
args		Args that will be passed to the coroutine	()
kwargs		Kwargs that will be passed to the coroutine	{}

Returns:

Type	Description
Union[Coroutine, Awaitable]	The coroutine instance which can be used in an await expression or asyncio functions

Source code in openhivenpy\events\__init__.py

def __call__(self, *args, **kwargs) -> Union[Coroutine, Awaitable]:
    """
    Returns the dispatch function of the class itself

    :param event_data: Data of the received event
    :param args: Args that will be passed to the coroutine
    :param kwargs: Kwargs that will be passed to the coroutine
    :return: The coroutine instance which can be used in an await
     expression or asyncio functions
    """
    dispatch: Union[Callable, Union[Awaitable, Callable]] = getattr(
        self, 'dispatch'
    )
    return dispatch(*args, **kwargs)

__init__(self, client, event_name, awaitable) special

Source code in openhivenpy\events\__init__.py

def __init__(
        self,
        client: HivenClient,
        event_name: str,
        awaitable: Union[Awaitable, Callable]
):
    self._client = client
    self._event_name = event_name
    self._awaitable: Optional[Awaitable] = None
    self.set_awaitable(awaitable)
    self._client.add_listener(self)

__repr__(self) special

Source code in openhivenpy\events\__init__.py

def __repr__(self):
    info = [
        ('event_name', getattr(self, 'event_name', None)),
        ('awaitable', getattr(self, 'awaitable', None))
    ]
    return '<MultiDispatchEventListener {}>'.format(
        ' '.join('%s=%s' % t for t in info)
    )

__str__(self) special

Source code in openhivenpy\events\__init__.py

def __str__(self):
    return f"<{self.__class__.__name__} for event {self.event_name}>"

dispatch(self) async

Source code in openhivenpy\events\__init__.py

async def dispatch(self) -> None:
    ...

set_awaitable(self, awaitable)

Sets the coroutine of the event_listener, which will be called when ws-events are received

Parameters:

Name	Type	Description	Default
awaitable	Union[Awaitable, Callable]	Awaitable (Coroutine Function). Can NOT be a initialised coroutine, since parameters will be passed on Runtime and coroutines are limited to a one-time execution. Coroutine functions or standard asyncio awaitable functions are defined using the async def syntax	required

Source code in openhivenpy\events\__init__.py

def set_awaitable(self, awaitable: Union[Awaitable, Callable]) -> None:
    """
    Sets the coroutine of the event_listener, which will be called when
    ws-events are received

    :param awaitable: Awaitable (Coroutine Function). Can NOT be a
     initialised coroutine, since parameters will be passed on Runtime and
     coroutines are limited to a one-time execution. Coroutine functions
     or standard asyncio awaitable functions are defined using the
     `async def` syntax
    """
    if inspect.isawaitable(awaitable) \
            or inspect.iscoroutinefunction(awaitable):
        self._awaitable = awaitable
    else:
        raise RuntimeError(
            f"Expected awaitable, but got {type(awaitable)}"
        )

Important

Usage classes, which represent DispatchEventListener

openhivenpy.events.MultiDispatchEventListener

EventListener Class that is used primarily for EventListeners that will be called multiple times

Methods

dispatch(self, *args, **kwargs) async

Dispatches the EventListener and calls a coroutine if one was passed. Does not raise exceptions but silences them!

Parameters:

Name	Type	Description	Default
args		Args that will be passed to the coroutine	()
kwargs		Kwargs that will be passed to the coroutine	{}

Source code in openhivenpy\events\__init__.py

async def dispatch(self, *args, **kwargs) -> None:
    """
    Dispatches the EventListener and calls a coroutine if one was passed.
    Does not raise exceptions but silences them!

    :param args: Args that will be passed to the coroutine
    :param kwargs: Kwargs that will be passed to the coroutine
    """
    try:
        await self.awaitable(*args, **kwargs)
    except Exception as e:
        utils.log_traceback(
            brief=f"[EVENTS] Ignoring exception in {repr(self)}:",
            exc_info=sys.exc_info()
        )
        raise RuntimeError(
            f"Failed to execute assigned coroutine "
            f"'{self.awaitable.__name__}'"
        ) from e

openhivenpy.events.SingleDispatchEventListener

EventListener Class that will be called only once and will store the events data, args and kwargs

args: tuple property readonly

dispatched: bool property readonly

kwargs: dict property readonly

Methods

__init__(self, client, event_name, awaitable) special

Source code in openhivenpy\events\__init__.py

def __init__(self, client: HivenClient, event_name: str,
             awaitable: Union[Awaitable, Callable]):
    self._dispatched = False
    self._event_data = None
    self._args = None
    self._kwargs = None
    super().__init__(client, event_name, awaitable)

__repr__(self) special

Source code in openhivenpy\events\__init__.py

def __repr__(self):
    info = [
        ('event_name', getattr(self, 'event_name', None)),
        ('dispatched', self.dispatched),
        ('awaitable', getattr(self, 'awaitable', None))
    ]
    return '<SingleDispatchEventListener {}>'.format(
        ' '.join('%s=%s' % t for t in info)
    )

dispatch(self, *args, **kwargs) async

Dispatches the EventListener and calls a coroutine if one was passed

Parameters:

Name	Type	Description	Default
args		Args that will be passed to the coroutine	()
kwargs		Kwargs that will be passed to the coroutine	{}

Returns:

Type	Description
None	Returns the passed event_data

Source code in openhivenpy\events\__init__.py

async def dispatch(self, *args, **kwargs) -> None:
    """
    Dispatches the EventListener and calls a coroutine if one was passed

    :param args: Args that will be passed to the coroutine
    :param kwargs: Kwargs that will be passed to the coroutine
    :return: Returns the passed event_data
    """
    try:
        self._args = args
        self._kwargs = kwargs
        await self.awaitable(*args, **kwargs)

    except Exception as e:
        utils.log_traceback(
            brief=f"[EVENTS] Ignoring exception in {repr(self)}:",
            exc_info=sys.exc_info()
        )
        raise RuntimeError(
            f"Failed to execute assigned coroutine "
            f"'{self.awaitable.__name__}'"
        ) from e
    self._dispatched = True
    self.client.remove_listener(self)

Important

The HivenEventHandler is already inherited in the standard client (HivenClient), so all properties and methods can be used there as well.

openhivenpy.events.HivenEventHandler

Events class used to register the main event listeners. Is inherited by the HivenClient for easier access.

active_listeners: Dict[str, List[DispatchEventListener]] property readonly

available_events: List[str] property readonly

non_buffer_events: List[str] property readonly

Methods

__init__(self, client, parsers) special

Source code in openhivenpy\events\__init__.py

def __init__(self, client: HivenClient, parsers: HivenParsers):
    self.parsers = parsers
    self._client = client
    self._active_listeners = {}
    self._available_events = EVENTS
    self._non_buffer_events = NON_BUFFER_EVENTS

    # Searching through the HivenClient to find all async functions that
    # were registered for event_listening Regular functions will NOT be
    # registered and only if they are async! This will avoid that errors
    # are thrown when trying to call the functions using 'await'
    for listener in inspect.getmembers(
            self, predicate=inspect.iscoroutinefunction
    ):
        func_name = listener[0].replace('on_', '')
        awaitable = listener[1]
        if func_name in self._available_events:
            self.add_multi_listener(func_name, awaitable)
            logger.debug(f"[EVENTS] Event {listener[0]} registered")

add_listener(self, listener)

Adds the listener to the client cache and will create a new list if the event_name does not exist yet!

Parameters:

Name	Type	Description	Default
listener	DispatchEventListener	The Listener that will be added	required

Source code in openhivenpy\events\__init__.py

def add_listener(self, listener: DispatchEventListener) -> None:
    """
    Adds the listener to the client cache and will create a new list if
     the event_name does not exist yet!

    :param listener: The Listener that will be added
    """
    if self.active_listeners.get(listener.event_name):
        self.active_listeners[listener.event_name].append(listener)
    else:
        self.active_listeners[listener.event_name] = [listener]

add_multi_listener(self, event_name, awaitable)

Adds a new event listener to the list of active listeners

Parameters:

Name	Type	Description	Default
event_name	str	The key/name of the event the EventListener should be listening to	required
awaitable	Union[Callable, Awaitable]	Coroutine that should be called when the EventListener was dispatched	required

Returns:

Type	Description
MultiDispatchEventListener	The newly created EventListener

Source code in openhivenpy\events\__init__.py

def add_multi_listener(
        self,
        event_name: str,
        awaitable: Union[Callable, Awaitable]
) -> MultiDispatchEventListener:
    """
    Adds a new event listener to the list of active listeners

    :param event_name: The key/name of the event the EventListener should
      be listening to
    :param awaitable: Coroutine that should be called when the
     EventListener was dispatched
    :return: The newly created EventListener
    """
    event_name = event_name.replace('on_', '')
    if event_name not in self.available_events:
        raise UnknownEventError(
            "The passed event type is invalid/does not exist"
        )

    if self._active_listeners.get(event_name) is None:
        self._active_listeners[event_name] = []

    return MultiDispatchEventListener(self._client, event_name, awaitable)

add_single_listener(self, event_name, awaitable)

Adds a new single dispatch event listener to the list of active listeners

Parameters:

Name	Type	Description	Default
event_name	str	The key/name of the event the EventListener should be listening to	required
awaitable	Union[Callable, Coroutine]	Coroutine that should be called when the EventListener was dispatched	required

Returns:

Type	Description
SingleDispatchEventListener	The newly created EventListener

Source code in openhivenpy\events\__init__.py

def add_single_listener(
        self,
        event_name: str,
        awaitable: Union[Callable, Coroutine]
) -> SingleDispatchEventListener:
    """
    Adds a new single dispatch event listener to the list of active
    listeners

    :param event_name: The key/name of the event the EventListener should
     be listening to
    :param awaitable: Coroutine that should be called when the
     EventListener was dispatched
    :return: The newly created EventListener
    """
    event_name = event_name.replace('on_', '')
    if event_name not in self.available_events:
        raise UnknownEventError(
            "The passed event type is invalid/does not exist"
        )

    if self._active_listeners.get(event_name) is None:
        self._active_listeners[event_name] = []

    return SingleDispatchEventListener(self._client, event_name, awaitable)

call_listeners(self, event_name, args, kwargs) async

Dispatches all active EventListeners for the specified event. Does not call the parsers but the function directly and requires the args, kwargs passed

Will run all tasks before returning! Only supposed to be called in cases of special events!

Parameters:

Name	Type	Description	Default
event_name	str	The name of the event that should be triggered	required
args	tuple	Args that will be passed to the coroutines	required
kwargs	dict	Kwargs that will be passed to the coroutines	required

Source code in openhivenpy\events\__init__.py

async def call_listeners(
        self, event_name: str, args: tuple, kwargs: dict
) -> None:
    """
    Dispatches all active EventListeners for the specified event.
    Does not call the parsers but the function directly and requires the
    args, kwargs passed

    Will run all tasks before returning! Only supposed to be called in
    cases of special events!

    :param event_name: The name of the event that should be triggered
    :param args: Args that will be passed to the coroutines
    :param kwargs: Kwargs that will be passed to the coroutines
    """
    listeners: List[DispatchEventListener] = self._active_listeners.get(
        event_name.lower().replace('on_', '')
    )
    if listeners:
        tasks = [listener(*args, **kwargs) for listener in listeners]
        await asyncio.gather(*tasks)

cleanup_listeners(self)

Cleanups the listeners and empties all active listeners

Source code in openhivenpy\events\__init__.py

def cleanup_listeners(self) -> None:
    """ Cleanups the listeners and empties all active listeners """
    self._active_listeners = {}

dispatch_event(self, event_name, args, kwargs)

Manually adds an event to the event_buffer and triggers all listeners. Will return immediately and does not require asyncio unlike call_listeners which only calls the listeners;

Note that this will NOT call the event parsers but manually directly pass the args and kwargs to the functions!

Parameters:

Name	Type	Description	Default
event_name	str	The name of the event that should be triggered	required
args	tuple	Args that will be passed to the coroutines	required
kwargs	dict	Kwargs that will be passed to the coroutines	required

Source code in openhivenpy\events\__init__.py

def dispatch_event(
        self, event_name: str, args: tuple, kwargs: dict
) -> None:
    """
    Manually adds an event to the event_buffer and triggers all listeners.
    Will return immediately and does not require asyncio unlike
    call_listeners which only calls the listeners;

    Note that this will NOT call the event parsers but manually directly
    pass the args and kwargs to the functions!

    :param event_name: The name of the event that should be triggered
    :param args: Args that will be passed to the coroutines
    :param kwargs: Kwargs that will be passed to the coroutines
    """
    self._validate_existence_of_event(event_name)

    _: MessageBroker = getattr(self, 'message_broker')
    buffer = _.get_buffer(event_name)
    buffer.add_new_event(
        {
            "callee": event_name,
            "caller": "client",
            "time": time.time()
        },
        args,
        kwargs
    )

event(self, awaitable=None)

Decorator used for registering Client Events

Parameters:

Name	Type	Description	Default
awaitable	Union[Callable, Coroutine]	Function that should be wrapped and registered	None

Source code in openhivenpy\events\__init__.py

def event(self, awaitable: Union[Callable, Coroutine] = None) -> Callable:
    """
    Decorator used for registering Client Events

    :param awaitable: Function that should be wrapped and registered
    """

    def decorator(awaitable: Union[Callable, Coroutine]) -> Callable:
        if not inspect.iscoroutinefunction(awaitable):
            raise TypeError(
                f"A coroutine was expected, got {type(awaitable)}"
            )

        self._validate_existence_of_event(awaitable.__name__)

        func_name = awaitable.__name__.replace('on_', '')
        self.add_multi_listener(func_name, awaitable)
        logger.debug(f"[EVENTS] Event {func_name} registered")

        # func can still be used normally outside the event listening
        # process
        return awaitable

    if awaitable is None:
        return decorator
    else:
        return decorator(awaitable)

remove_listener(self, listener)

Removes the listener from the client cache

Parameters:

Name	Type	Description	Default
listener	DispatchEventListener	The Listener that will be removed	required

Exceptions:

Type	Description
KeyError	If the listener does not exist in the cache

Source code in openhivenpy\events\__init__.py

def remove_listener(self, listener: DispatchEventListener) -> None:
    """
    Removes the listener from the client cache

    :param listener: The Listener that will be removed
    :raises KeyError: If the listener does not exist in the cache
    """
    if self.active_listeners.get(listener.event_name):
        self.active_listeners[listener.event_name].remove(listener)
    else:
        raise KeyError("The listener does not exist in the cache")

wait_for(self, event_name, awaitable=None) async

Waits for an event to be triggered and then returns the args and *kwargs passed

Parameters:

Name	Type	Description	Default
event_name	str	Name of the event to wait for	required
awaitable	Union[Callable, Coroutine, None]	Coroutine that can be passed to be additionally triggered when received	None

Returns:

Type	Description
Tuple[tuple, dict]	A tuple of the args and kwargs => (args, kwargs)

Exceptions:

Type	Description
UnknownEventError	If the event does not exist

Source code in openhivenpy\events\__init__.py

async def wait_for(
        self,
        event_name: str,
        awaitable: Union[Callable, Coroutine, None] = None
) -> Tuple[tuple, dict]:
    """
    Waits for an event to be triggered and then returns the *args and
    **kwargs passed

    :param event_name: Name of the event to wait for
    :param awaitable: Coroutine that can be passed to be additionally
     triggered when received
    :raises UnknownEventError: If the event does not exist
    :return: A tuple of the args and kwargs => (args, kwargs)
    """
    event_name = event_name.replace('on_', '')
    if event_name not in self.available_events:
        raise UnknownEventError(
            "The passed event type is invalid/does not exist"
        )

    if awaitable is None:
        async def _empty(*args, **kwargs):
            ...

        awaitable = _empty

    listener = self.add_single_listener(event_name, awaitable)
    while not listener.dispatched:
        await asyncio.sleep(.05)

    return listener.args, listener.kwargs

---

Last update: 2021-08-22