# -*- 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
[docs]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
[docs] def app(self) -> traits.RESTAware:
"""App instance for this application."""
@classmethod
[docs] def dispatches(cls) -> typing.Sequence[typing.Type[Event]]:
"""Sequence of the event classes this event is dispatched as."""
return cls.__dispatches
@classmethod
[docs] def bitmask(cls) -> int:
"""Bitmask for this event."""
return cls.__bitmask
[docs]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
[docs]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
[docs]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
[docs]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)
[docs]class ExceptionEvent(Event, typing.Generic[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.
"""
[docs] exception: Exception = attr.field()
"""Exception that was raised."""
[docs] failed_event: EventT = attr.field()
"""Event instance that caused the exception."""
[docs] failed_callback: FailedCallbackT[EventT] = attr.field()
"""Event callback that threw an exception."""
@property
[docs] def app(self) -> traits.RESTAware:
# <<inherited docstring from Event>>.
return self.failed_event.app
@property
[docs] def shard(self) -> typing.Optional[gateway_shard.GatewayShard]:
"""Shard that received the event, if there was one associated.
This may be `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
[docs] 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`.
The `sys.exc_info` tiplet consists of the exception type, the exception
instance, and the traceback of the exception.
"""
return type(self.exception), self.exception, self.exception.__traceback__
[docs] 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)