Common#
Common reference#
- class nextcore.common.Dispatcher#
A event dispatcher
A event dispatcher has 2 different ways to listen to events. - Event handlers - Wait for
Event handlers are callbacks that are called when an event is dispatched.
Wait for is a way to wait for an event to be dispatched inside a async function. They are temporary listeners that gets removed once the event is dispatched.
There is also “global” events. This makes all handlers get every event. This will also provide a
EventNameT
parameter to the handler (and check if you are usingDispatcher.wait_for()
).Example usage:
dispatcher: Dispatcher[str] = Dispatcher() @dispatcher.listen("join") async def join_handler(username: str) -> None: print(f"Welcome {username}") await dispatcher.dispatch("join", "John")
- add_error_handler(callback: ExceptionHandler, event_name: EventNameT) None #
- add_error_handler(callback: GlobalExceptionHandler[EventNameT], event_name: None = None) None
Add an error handler for listeners.
The callback will be called when an exception is raised in a listener.
Note
If no listener is registered, the error will be logged.
Example usage:
async def error(exception: Exception) -> None: print("Oops!") dispatcher.add_error_handler(error_handler, "join")
- Parameters:
callback – The callback to be called whenever a error occurs
event_name –
The event to restrict the error handler to.
Note
Passing this will make callback not be given the event name.
- add_listener(callback: EventCallback, event_name: EventNameT) None #
- add_listener(callback: GlobalEventCallback[EventNameT], event_name: None = None) None
Add a event listener.
Example usage:
async def join_handler(username: str) -> None: print(f"Welcome {username}") dispatcher.add_listener(welcome_handler, "join")
- Parameters:
callback – The event callback to register.
event_name – The event name to listen to. If this is
None
, this is considered a global event and all events will be sent to the callback.
- close()#
Clean up internal state and listeners.
- Return type:
None
- async dispatch(event_name, *args)#
Dispatch a event
Example usage:
dispatcher.dispatch("join", "John")
- Parameters:
event_name (EventNameT) – The event name to dispatch to.
args (Any) – The event arguments. This will be passed to the listeners.
- Return type:
None
- listen(event_name: EventNameT) Callable[[EventCallbackT], EventCallbackT] #
- listen(event_name: None = None) Callable[[GlobalEventCallback[EventNameT]], GlobalEventCallback[EventNameT]]
Decorator to register a event listener.
Example usage:
@dispatcher.listen("join") async def join_handler(username: str) -> None: print(f"Welcome {username}")
- Parameters:
event_name – The event name to register the listener to. If this is
None
, the listener is considered a global event.
- remove_error_handler(callback: ExceptionHandler, event_name: EventNameT) None #
- remove_error_handler(callback: GlobalExceptionHandler[EventNameT], event_name: None = None) None
Removes an error handler.
Example usage:
dispatcher.remove_error_handler(error_handler, "join")
- Parameters:
callback – The error handler to remove.
event_name –
The event name to remove. If this is
None
, the listener is considered a global event.Warning
If the event_name does not match the event name it was registered with, the removal will fail with a
ValueError
as the listener was not found.
- Raises:
ValueError – The callback was not registered to this event.
- remove_listener(callback: EventCallback, event_name: EventNameT) None #
- remove_listener(callback: GlobalEventCallback[EventNameT], event_name: None = None) None
Removes a event listener.
Example usage:
dispatcher.remove_listener(welcome_handler, "join")
- Parameters:
callback – The event callback to remove.
event_name –
The event name to remove. If this is
None
, the listener is considered a global event.Warning
If the event_name does not match the event name it was registered with, the removal will fail with a
ValueError
as the listener was not found.
- Raises:
ValueError – The callback was not registered to this event.
- async wait_for(check: WaitForCheck, event_name: EventNameT) WaitForReturn #
- async wait_for(check: GlobalWaitForCheck[EventNameT], event_name: None = None) GlobalWaitForReturn[EventNameT]
Wait for an event to occur.
Example usage:
username = await dispatcher.wait_for( lambda username: username.startswith("h"), "join" ) print(f"{username}'s username starts with h")
- Parameters:
check – Check for it to return.
event_name – The event name to wait for. If this is
None
, it is considered global and will return when any event is received and include aEventNameT
as the first event argument.
- Returns:
EventNameT
, *typing.Any
– The event name and the event arguments. This is only returned if the event is global.*
typing.Any
– The event arguments.
- class nextcore.common.TimesPer(limit, per)#
A smart TimesPer implementation.
- Parameters:
- limit#
The amount of times the rate limiter can be used
- per#
How often this resets in seconds
- reset_offset_seconds#
How much the resetting should be offset to account for processing/networking delays.
This will be added to the reset time, so for example a offset of
1
will make resetting 1 second slower.
- acquire(*, priority=0, wait=True)#
Use a spot in the rate-limit.
- Parameters:
- Raises:
RateLimitedError –
wait
was set toFalse
and there was no more spots in the rate limit.- Returns:
A context manager that will wait in __aenter__ until a request should be made.
- Return type:
- async close()#
Cleanup this instance.
This should be done when this instance is never going to be used anymore
Warning
Continued use of this instance will result in instability
- Return type:
None
- class nextcore.common.UndefinedType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)#
A second
None
for specifying that it should not be provided.Example usage:
from nextcore.common import UndefinedType, UNDEFINED thing = UNDEFINED if thing is UNDEFINED: print("Thing is undefined!")
- nextcore.common.Undefined: UndefinedType#
A alias for
UndefinedType.UNDEFINED