Base types and functions for events in Hikari.

This module

Expand source code
Browse git
# -*- coding: utf-8 -*-
# cython: language_level=3
# Copyright (c) 2020 Nekokatt
# Copyright (c) 2021-present davfsa
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Base types and functions for events in Hikari."""
from __future__ import annotations

__all__: typing.Sequence[str] = (
    "Event",
    "ExceptionEvent",
    "EventT",
    "is_no_recursive_throw_event",
    "no_recursive_throw",
    "get_required_intents_for",
    "requires_intents",
)

import abc
import inspect
import typing

import attr

from hikari import intents
from hikari import traits
from hikari.api import shard as gateway_shard
from hikari.internal import attr_extensions

if typing.TYPE_CHECKING:
    import types

    _T = typing.TypeVar("_T")

REQUIRED_INTENTS_ATTR: typing.Final[str] = "___requiresintents___"
NO_RECURSIVE_THROW_ATTR: typing.Final[str] = "___norecursivethrow___"

_id_counter = 1  # We start at 1 since Event is 0


class Event(abc.ABC):
    """Base event type that all Hikari events should subclass."""

    __slots__: typing.Sequence[str] = ()

    __dispatches: typing.ClassVar[typing.Tuple[typing.Type[Event], ...]]
    __bitmask: typing.ClassVar[int]

    def __init_subclass__(cls) -> None:
        super().__init_subclass__()
        # hasattr doesn't work with private variables in this case so we use a try except.
        # We need to set Event's __dispatches when the first subclass is made as Event cannot
        # be included in a tuple literal on itself due to not existing yet.
        try:
            Event.__dispatches
        except AttributeError:
            Event.__dispatches = (Event,)
            Event.__bitmask = 1 << 0

        global _id_counter

        mro = cls.mro()
        # We don't have to explicitly include Event here as issubclass(Event, Event) returns True.
        # Non-event classes should be ignored.
        cls.__dispatches = tuple(sub_cls for sub_cls in mro if issubclass(sub_cls, Event))
        cls.__bitmask = 1 << _id_counter
        _id_counter += 1

    @property
    @abc.abstractmethod
    def app(self) -> traits.RESTAware:
        """App instance for this application.

        Returns
        -------
        hikari.traits.RESTAware
            The REST-aware app trait.
        """

    @classmethod
    def dispatches(cls) -> typing.Sequence[typing.Type[Event]]:
        """Sequence of the event classes this event is dispatched as."""
        return cls.__dispatches

    @classmethod
    def bitmask(cls) -> int:
        """Bitmask for this event."""
        return cls.__bitmask


def get_required_intents_for(event_type: typing.Type[Event]) -> typing.Collection[intents.Intents]:
    """Retrieve the intents that are required to listen to an event type.

    Parameters
    ----------
    event_type : typing.Type[Event]
        The event type to get required intents for.

    Returns
    -------
    typing.Collection[hikari.intents.Intents]
        Collection of acceptable subset combinations of intent needed to
        be able to receive the given event type.
    """
    result = getattr(event_type, REQUIRED_INTENTS_ATTR, ())
    assert isinstance(result, typing.Collection)
    return result


def requires_intents(first: intents.Intents, *rest: intents.Intents) -> typing.Callable[[_T], _T]:
    """Decorate an event type to define what intents it requires.

    Parameters
    ----------
    first : hikari.intents.Intents
        First combination of intents that are acceptable in order to receive
        the decorated event type.
    *rest : hikari.intents.Intents
        Zero or more additional combinations of intents to require for this
        event to be subscribed to.
    """

    def decorator(cls: _T) -> _T:
        required_intents = [first, *rest]
        setattr(cls, REQUIRED_INTENTS_ATTR, required_intents)
        doc = inspect.getdoc(cls) or ""
        doc += "\n\nThis requires one of the following combinations of intents in order to be dispatched:\n\n"
        for intent_group in required_intents:
            preview = " + ".join(
                f"`{type(i).__module__}.{type(i).__qualname__}.{i.name}`" for i in intent_group.split()
            )
            doc += f" - {preview}\n"

        cls.__doc__ = doc
        return cls

    return decorator


def no_recursive_throw() -> typing.Callable[[_T], _T]:
    """Decorate an event type to indicate errors should not be handled.

    This is useful for exception event types that you do not want to
    have invoked recursively.
    """

    def decorator(cls: _T) -> _T:
        setattr(cls, NO_RECURSIVE_THROW_ATTR, True)
        doc = inspect.getdoc(cls) or ""
        doc += (
            "\n"
            "!!! warning\n"
            "    Any exceptions raised by handlers for this event will be dumped to the\n"
            "    application logger and silently discarded, preventing recursive loops\n"
            "    produced by faulty exception event handling. Thus, it is imperative\n"
            "    that you ensure any exceptions are explicitly caught within handlers\n"
            "    for this event if they may occur.\n"
        )
        cls.__doc__ = doc
        return cls

    return decorator


def is_no_recursive_throw_event(obj: typing.Union[_T, typing.Type[_T]]) -> bool:
    """Return True if this event is marked as `___norecursivethrow___`."""
    result = getattr(obj, NO_RECURSIVE_THROW_ATTR, False)
    assert isinstance(result, bool)
    return result


EventT = typing.TypeVar("EventT", bound=Event)
FailedCallbackT = typing.Callable[[EventT], typing.Coroutine[typing.Any, typing.Any, None]]


@no_recursive_throw()
@attr_extensions.with_copy
@attr.define(kw_only=True, weakref_slot=False)
class ExceptionEvent(Event, typing.Generic[EventT]):
    """Event that is raised when another event handler raises an `Exception`.

    !!! note
        Only exceptions that derive from `builtins.Exception` will be caught.
        Other exceptions outside this range will propagate past this callback.
        This prevents event handlers interfering with critical exceptions
        such as `KeyboardError` which would have potentially undesired
        side-effects on the application runtime.
    """

    exception: Exception = attr.field()
    """Exception that was raised.

    Returns
    -------
    builtins.Exception
        Exception that was raised in the event handler.
    """

    failed_event: EventT = attr.field()
    """Event instance that caused the exception.

    Returns
    -------
    hikari.events.base_events.Event
        Event that was being processed when the exception occurred.
    """

    failed_callback: FailedCallbackT[EventT] = attr.field()
    """Event callback that threw an exception.

    Returns
    -------
    callback
        Event callback that failed execution.
    """

    @property
    def app(self) -> traits.RESTAware:
        # <<inherited docstring from Event>>.
        return self.failed_event.app

    @property
    def shard(self) -> typing.Optional[gateway_shard.GatewayShard]:
        """Shard that received the event, if there was one associated.

        Returns
        -------
        typing.Optional[hikari.api.shard.GatewayShard]
            Shard that raised this exception.

            This may be `builtins.None` if no specific shard was the cause of this
            exception (e.g. when starting up or shutting down).
        """
        shard = getattr(self.failed_event, "shard", None)
        if isinstance(shard, gateway_shard.GatewayShard):
            return shard
        return None

    @property
    def exc_info(self) -> typing.Tuple[typing.Type[Exception], Exception, typing.Optional[types.TracebackType]]:
        """Exception triplet that follows the same format as `sys.exc_info`.

        Returns
        -------
        builtins.tuple[typing.Type[Exception], Exception, typing.Optional[types.TracebackType]]
            The `sys.exc_info`-compatible tuple of the exception type, the
            exception instance, and the traceback of the exception.
        """
        return type(self.exception), self.exception, self.exception.__traceback__

    async def retry(self) -> None:
        """Invoke the failed event again.

        If an exception is thrown this time, it will need to be manually
        caught in-code, or will be discarded.
        """
        await self.failed_callback(self.failed_event)

Functions

def get_required_intents_for(
    event_type: Type[Event],
) -> Collection[Intents]: ...

Retrieve the intents that are required to listen to an event type.

Parameters

event_type : Type[Event]
The event type to get required intents for.

Returns

Collection[Intents]
Collection of acceptable subset combinations of intent needed to be able to receive the given event type.
Expand source code
Browse git
def get_required_intents_for(event_type: typing.Type[Event]) -> typing.Collection[intents.Intents]:
    """Retrieve the intents that are required to listen to an event type.

    Parameters
    ----------
    event_type : typing.Type[Event]
        The event type to get required intents for.

    Returns
    -------
    typing.Collection[hikari.intents.Intents]
        Collection of acceptable subset combinations of intent needed to
        be able to receive the given event type.
    """
    result = getattr(event_type, REQUIRED_INTENTS_ATTR, ())
    assert isinstance(result, typing.Collection)
    return result
def is_no_recursive_throw_event(
    obj: Union[_T, Type[_T]],
) -> bool: ...

Return True if this event is marked as ___norecursivethrow___.

Expand source code
Browse git
def is_no_recursive_throw_event(obj: typing.Union[_T, typing.Type[_T]]) -> bool:
    """Return True if this event is marked as `___norecursivethrow___`."""
    result = getattr(obj, NO_RECURSIVE_THROW_ATTR, False)
    assert isinstance(result, bool)
    return result
def no_recursive_throw() -> Callable[[_T], _T]: ...

Decorate an event type to indicate errors should not be handled.

This is useful for exception event types that you do not want to have invoked recursively.

Expand source code
Browse git
def no_recursive_throw() -> typing.Callable[[_T], _T]:
    """Decorate an event type to indicate errors should not be handled.

    This is useful for exception event types that you do not want to
    have invoked recursively.
    """

    def decorator(cls: _T) -> _T:
        setattr(cls, NO_RECURSIVE_THROW_ATTR, True)
        doc = inspect.getdoc(cls) or ""
        doc += (
            "\n"
            "!!! warning\n"
            "    Any exceptions raised by handlers for this event will be dumped to the\n"
            "    application logger and silently discarded, preventing recursive loops\n"
            "    produced by faulty exception event handling. Thus, it is imperative\n"
            "    that you ensure any exceptions are explicitly caught within handlers\n"
            "    for this event if they may occur.\n"
        )
        cls.__doc__ = doc
        return cls

    return decorator
def requires_intents(
    first: intents.Intents,
    *rest: intents.Intents,
) -> Callable[[_T], _T]: ...

Decorate an event type to define what intents it requires.

Parameters

first : Intents
First combination of intents that are acceptable in order to receive the decorated event type.
*rest : Intents
Zero or more additional combinations of intents to require for this event to be subscribed to.
Expand source code
Browse git
def requires_intents(first: intents.Intents, *rest: intents.Intents) -> typing.Callable[[_T], _T]:
    """Decorate an event type to define what intents it requires.

    Parameters
    ----------
    first : hikari.intents.Intents
        First combination of intents that are acceptable in order to receive
        the decorated event type.
    *rest : hikari.intents.Intents
        Zero or more additional combinations of intents to require for this
        event to be subscribed to.
    """

    def decorator(cls: _T) -> _T:
        required_intents = [first, *rest]
        setattr(cls, REQUIRED_INTENTS_ATTR, required_intents)
        doc = inspect.getdoc(cls) or ""
        doc += "\n\nThis requires one of the following combinations of intents in order to be dispatched:\n\n"
        for intent_group in required_intents:
            preview = " + ".join(
                f"`{type(i).__module__}.{type(i).__qualname__}.{i.name}`" for i in intent_group.split()
            )
            doc += f" - {preview}\n"

        cls.__doc__ = doc
        return cls

    return decorator

Classes

class Event

class Event: ...

Base event type that all Hikari events should subclass.

Expand source code
Browse git
class Event(abc.ABC):
    """Base event type that all Hikari events should subclass."""

    __slots__: typing.Sequence[str] = ()

    __dispatches: typing.ClassVar[typing.Tuple[typing.Type[Event], ...]]
    __bitmask: typing.ClassVar[int]

    def __init_subclass__(cls) -> None:
        super().__init_subclass__()
        # hasattr doesn't work with private variables in this case so we use a try except.
        # We need to set Event's __dispatches when the first subclass is made as Event cannot
        # be included in a tuple literal on itself due to not existing yet.
        try:
            Event.__dispatches
        except AttributeError:
            Event.__dispatches = (Event,)
            Event.__bitmask = 1 << 0

        global _id_counter

        mro = cls.mro()
        # We don't have to explicitly include Event here as issubclass(Event, Event) returns True.
        # Non-event classes should be ignored.
        cls.__dispatches = tuple(sub_cls for sub_cls in mro if issubclass(sub_cls, Event))
        cls.__bitmask = 1 << _id_counter
        _id_counter += 1

    @property
    @abc.abstractmethod
    def app(self) -> traits.RESTAware:
        """App instance for this application.

        Returns
        -------
        hikari.traits.RESTAware
            The REST-aware app trait.
        """

    @classmethod
    def dispatches(cls) -> typing.Sequence[typing.Type[Event]]:
        """Sequence of the event classes this event is dispatched as."""
        return cls.__dispatches

    @classmethod
    def bitmask(cls) -> int:
        """Bitmask for this event."""
        return cls.__bitmask
Subclasses
dataclass ExceptionEvent

Event that is raised when another event handler raises an Exception

dataclass StartedEvent

Event that is triggered after the application has started …

dataclass StartingEvent

Event that is triggered before the application connects to discord …

dataclass StoppedEvent

Event that is triggered once the application has disconnected …

dataclass StoppingEvent

Event that is triggered as soon as the application is requested to close …

abstract class ShardEvent

Base class for any event that was shard-specific.

Method resolution order
abstract class Event
That's this class!
extern class abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

Variables and properties
abstract property appRESTAware

App instance for this application.

Returns

RESTAware
The REST-aware app trait.
Methods
def bitmask() -> int: ...

Bitmask for this event.

Expand source code
Browse git
@classmethod
def bitmask(cls) -> int:
    """Bitmask for this event."""
    return cls.__bitmask
def dispatches() -> Sequence[Type[Event]]: ...

Sequence of the event classes this event is dispatched as.

Expand source code
Browse git
@classmethod
def dispatches(cls) -> typing.Sequence[typing.Type[Event]]:
    """Sequence of the event classes this event is dispatched as."""
    return cls.__dispatches

dataclass ExceptionEvent

class ExceptionEvent (
    *,
    exception: Exception,
    failed_event: EventT,
    failed_callback: FailedCallbackT[EventT],
): ...

Event that is raised when another event handler raises an Exception.

Note

Only exceptions that derive from Exception will be caught. Other exceptions outside this range will propagate past this callback. This prevents event handlers interfering with critical exceptions such as KeyboardError which would have potentially undesired side-effects on the application runtime.

Warning

Any exceptions raised by handlers for this event will be dumped to the application logger and silently discarded, preventing recursive loops produced by faulty exception event handling. Thus, it is imperative that you ensure any exceptions are explicitly caught within handlers for this event if they may occur.

Method generated by attrs for class ExceptionEvent.

Expand source code
Browse git
class ExceptionEvent(Event, typing.Generic[EventT]):
    """Event that is raised when another event handler raises an `Exception`.

    !!! note
        Only exceptions that derive from `builtins.Exception` will be caught.
        Other exceptions outside this range will propagate past this callback.
        This prevents event handlers interfering with critical exceptions
        such as `KeyboardError` which would have potentially undesired
        side-effects on the application runtime.
    """

    exception: Exception = attr.field()
    """Exception that was raised.

    Returns
    -------
    builtins.Exception
        Exception that was raised in the event handler.
    """

    failed_event: EventT = attr.field()
    """Event instance that caused the exception.

    Returns
    -------
    hikari.events.base_events.Event
        Event that was being processed when the exception occurred.
    """

    failed_callback: FailedCallbackT[EventT] = attr.field()
    """Event callback that threw an exception.

    Returns
    -------
    callback
        Event callback that failed execution.
    """

    @property
    def app(self) -> traits.RESTAware:
        # <<inherited docstring from Event>>.
        return self.failed_event.app

    @property
    def shard(self) -> typing.Optional[gateway_shard.GatewayShard]:
        """Shard that received the event, if there was one associated.

        Returns
        -------
        typing.Optional[hikari.api.shard.GatewayShard]
            Shard that raised this exception.

            This may be `builtins.None` if no specific shard was the cause of this
            exception (e.g. when starting up or shutting down).
        """
        shard = getattr(self.failed_event, "shard", None)
        if isinstance(shard, gateway_shard.GatewayShard):
            return shard
        return None

    @property
    def exc_info(self) -> typing.Tuple[typing.Type[Exception], Exception, typing.Optional[types.TracebackType]]:
        """Exception triplet that follows the same format as `sys.exc_info`.

        Returns
        -------
        builtins.tuple[typing.Type[Exception], Exception, typing.Optional[types.TracebackType]]
            The `sys.exc_info`-compatible tuple of the exception type, the
            exception instance, and the traceback of the exception.
        """
        return type(self.exception), self.exception, self.exception.__traceback__

    async def retry(self) -> None:
        """Invoke the failed event again.

        If an exception is thrown this time, it will need to be manually
        caught in-code, or will be discarded.
        """
        await self.failed_callback(self.failed_event)
Method resolution order
dataclass ExceptionEvent
That's this class!
abstract class Event

Base event type that all Hikari events should subclass.

extern class abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

extern class Generic

Abstract base class for generic types …

Variables and properties
property appRESTAware

App instance for this application.

Returns

RESTAware
The REST-aware app trait.
property exc_infoTuple[Type[Exception], ExceptionOptional[types.TracebackType]]

Exception triplet that follows the same format as sys.exc_info.

Returns

tuple[Type[Exception], Exception, Optional[types.TracebackType]]
The sys.exc_info-compatible tuple of the exception type, the exception instance, and the traceback of the exception.
property exceptionException

Exception that was raised.

Returns

Exception
Exception that was raised in the event handler.
property failed_callback : Callable[[~EventT], Coroutine[Any, Any, None]]

Event callback that threw an exception.

Returns

callback
Event callback that failed execution.
property failed_event : ~EventT

Event instance that caused the exception.

Returns

Event
Event that was being processed when the exception occurred.
property shard : Optional[GatewayShard]

Shard that received the event, if there was one associated.

Returns

Optional[GatewayShard]

Shard that raised this exception.

This may be None if no specific shard was the cause of this exception (e.g. when starting up or shutting down).

Methods
async def retry() -> None: ...

Invoke the failed event again.

If an exception is thrown this time, it will need to be manually caught in-code, or will be discarded.

Expand source code
Browse git
async def retry(self) -> None:
    """Invoke the failed event again.

    If an exception is thrown this time, it will need to be manually
    caught in-code, or will be discarded.
    """
    await self.failed_callback(self.failed_event)
def bitmask() -> int: ...

Inherited from: Event.bitmask

Bitmask for this event.

def dispatches() -> Sequence[Type[Event]]: ...

Inherited from: Event.dispatches

Sequence of the event classes this event is dispatched as.