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 using Dispatcher.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")
Return type

None

add_error_handler(callback: ExceptionHandler, event_name: nextcore.common.dispatcher.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: nextcore.common.dispatcher.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: nextcore.common.dispatcher.EventNameT) → Callable[[EventCallback], EventCallback]#
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: nextcore.common.dispatcher.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: nextcore.common.dispatcher.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: nextcore.common.dispatcher.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 a EventNameT 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 (int) – The amount of times the rate limiter can be used

  • per (float) – How often this resets in seconds

Return type

None

acquire(*, priority=0, wait=True)#

Use a spot in the rate-limit.

Parameters

  • priority (int) – The priority. Lower number means it will be requested earlier.

  • wait (bool) –

    Wait for a spot in the rate limit.

    If this is False, this will raise RateLimitedError instead.

Raises

RateLimitedError – wait was set to False 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

typing.AsyncContextManager

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

Errors#