Hooks

ballsdex.core.hooks

C

C = TypeVar('C', bound='HookContext')

HOOK_NAMES

HOOK_NAMES: frozenset[str] = frozenset((hook_name) for cls in (SpawnCheckContext, PreCatchContext, CatchRollContext, CatchMessageContext, PreTradeContext, PreGiveContext))

HOOK_TIMEOUT

HOOK_TIMEOUT = 3.0

HookFunc

HookFunc = Callable[[C], Awaitable[Any]]

__all__

__all__ = ('HOOK_TIMEOUT', 'HookContext', 'CancellableContext', 'SpawnCheckContext', 'PreCatchContext', 'CatchRollContext', 'CatchMessageContext', 'PreTradeContext', 'PreGiveContext', 'HookRegistry', 'hook')

log

log = getLogger('ballsdex.core.hooks')

CancellableContext

CancellableContext()

Bases: HookContext

A context that can be cancelled to block the action. Once cancelled, the remaining hooks are not called.

Attributes:

• cancelled (bool) –

  Whether a hook cancelled the action.

• reason (str | None) –

  The user-facing reason given when cancelling, if any.

cancelled

cancelled: bool = field(default=False, init=False)

hook_name

hook_name: str

reason

reason: str | None = field(default=None, init=False)

cancel

cancel(reason: str | None = None)

Block the action.

Parameters:

• reason (str | None, default: None ) –

  A message shown to the user. A generic message is shown if omitted.

Source code in ballsdex/core/hooks.py

def cancel(self, reason: str | None = None):
    """
    Block the action.

    Parameters
    ----------
    reason: str | None
        A message shown to the user. A generic message is shown if omitted.
    """
    self.cancelled = True
    self.reason = reason

CatchMessageContext

CatchMessageContext(*, view: BallSpawnView, interaction: Interaction, player: Player, ball_instance: BallInstance, is_new: bool, content: str)

Bases: HookContext

A countryball was caught, and the catch message is about to be sent.

Attributes:

• view (BallSpawnView) –

  The view of the spawn that was caught.

• interaction (Interaction) –

  The interaction of the catch prompt.

• player (Player) –

  The player who caught.

• ball_instance (BallInstance) –

  The countryball now owned by the player.

• is_new (bool) –

  Whether this is the first time the player owns this countryball.

• content (str) –

  The catch message that will be sent. Editable.

ball_instance

ball_instance: BallInstance

content

content: str

hook_name

hook_name = 'catch_message'

interaction

interaction: Interaction

is_new

is_new: bool

player

player: Player

view

view: BallSpawnView

CatchRollContext

CatchRollContext(*, view: BallSpawnView, user: User | Member, player: Player, special: Special | None, attack_bonus: int, health_bonus: int)

Bases: HookContext

A new countryball instance is about to be created after a catch. The rolled values can be modified. Not called when catching a dropped countryball, since it already exists.

Attributes:

• view (BallSpawnView) –

  The view of the spawn being caught.

• user (User | Member) –

  The user catching.

• player (Player) –

  The player catching.

• special (Special | None) –

  The special event applied. Editable.

• attack_bonus (int) –

  The attack bonus in percent. Editable.

• health_bonus (int) –

  The health bonus in percent. Editable.

attack_bonus

attack_bonus: int

health_bonus

health_bonus: int

hook_name

hook_name = 'catch_roll'

player

player: Player

special

special: Special | None

user

user: User | Member

view

view: BallSpawnView

HookContext

Base class for all hook contexts.

Attributes:

• hook_name (ClassVar[str]) –

  The name of the hook receiving this context.

hook_name

hook_name: str

HookRegistry

HookRegistry()

Holds all registered hooks. Available as bot.hooks.

Source code in ballsdex/core/hooks.py

def __init__(self):
    self._hooks: defaultdict[str, list[_RegisteredHook]] = defaultdict(list)

add_cog

add_cog(cog: Cog)

Register all methods of a cog decorated with hook.

Source code in ballsdex/core/hooks.py

def add_cog(self, cog: commands.Cog):
    """
    Register all methods of a cog decorated with [`hook`][ballsdex.core.hooks.hook].
    """
    seen: set[str] = set()
    for base in type(cog).__mro__:
        for attr, value in base.__dict__.items():
            if attr in seen:
                continue
            seen.add(attr)
            if marker := getattr(value, "__ballsdex_hook__", None):
                name, priority = marker
                self.register(name, getattr(cog, attr), priority=priority, owner=cog)

has

has(name: str) -> bool

Whether any function is registered for this hook.

Source code in ballsdex/core/hooks.py

def has(self, name: str) -> bool:
    """
    Whether any function is registered for this hook.
    """
    return bool(self._hooks.get(name))

register

register(name: str, func: HookFunc, *, priority: int = 0, owner: object | None = None)

Register a hook function.

Parameters:

• name (str) –

  The name of the hook.

• func (HookFunc) –

  The async function to call with the context.

• priority (int, default: 0 ) –

  Hooks with a higher priority run first. Defaults to 0.

• owner (object | None, default: None ) –

  The object owning this hook, used to unregister all its hooks at once.

Source code in ballsdex/core/hooks.py

def register(self, name: str, func: HookFunc, *, priority: int = 0, owner: object | None = None):
    """
    Register a hook function.

    Parameters
    ----------
    name: str
        The name of the hook.
    func: Callable[[HookContext], Awaitable[Any]]
        The async function to call with the context.
    priority: int
        Hooks with a higher priority run first. Defaults to 0.
    owner: object | None
        The object owning this hook, used to unregister all its hooks at once.
    """
    if name not in HOOK_NAMES:
        raise ValueError(f"Unknown hook {name!r}")
    if not inspect.iscoroutinefunction(func):
        raise TypeError("Hook functions must be coroutines")
    hooks = self._hooks[name]
    hooks.append(_RegisteredHook(func, priority, owner))
    hooks.sort(key=lambda h: h.priority, reverse=True)

remove_cog

remove_cog(cog: Cog)

Unregister all hooks of a cog.

Source code in ballsdex/core/hooks.py

def remove_cog(self, cog: commands.Cog):
    """
    Unregister all hooks of a cog.
    """
    self.unregister_owner(cog)

run

run(ctx: C) -> C

Run all functions registered for the hook of this context, and return the context.

Parameters:

• ctx (C) –

  The context passed to each hook function.

Source code in ballsdex/core/hooks.py

async def run(self, ctx: C) -> C:
    """
    Run all functions registered for the hook of this context, and return the context.

    Parameters
    ----------
    ctx: HookContext
        The context passed to each hook function.
    """
    hooks = self._hooks.get(ctx.hook_name)
    if not hooks:
        return ctx
    with tracing.span("hooks.run", resource=ctx.hook_name):
        for registered in list(hooks):
            try:
                await asyncio.wait_for(registered.func(ctx), timeout=HOOK_TIMEOUT)
            except asyncio.TimeoutError:
                log.warning(f"Hook {registered.func!r} for {ctx.hook_name} timed out, skipping.")
            except Exception:
                log.exception(f"Hook {registered.func!r} for {ctx.hook_name} raised an error, skipping.")
            if isinstance(ctx, CancellableContext) and ctx.cancelled:
                tracing.set_tag("hooks.cancelled", True)
                break
    return ctx

unregister

unregister(name: str, func: HookFunc)

Unregister a hook function. Does nothing if it wasn't registered.

Source code in ballsdex/core/hooks.py

def unregister(self, name: str, func: HookFunc):
    """
    Unregister a hook function. Does nothing if it wasn't registered.
    """
    self._hooks[name] = [h for h in self._hooks[name] if h.func != func]

unregister_owner

unregister_owner(owner: object)

Unregister all hook functions registered with this owner.

Source code in ballsdex/core/hooks.py

def unregister_owner(self, owner: object):
    """
    Unregister all hook functions registered with this owner.
    """
    for name, hooks in self._hooks.items():
        self._hooks[name] = [h for h in hooks if h.owner is not owner]

PreCatchContext

PreCatchContext(*, view: BallSpawnView, interaction: Interaction, player: Player, guess: str)

Bases: CancellableContext

A user guessed the name of a countryball correctly and is about to catch it. Cancelling prevents the catch, the countryball stays available to others.

Attributes:

• view (BallSpawnView) –

  The view of the spawn being caught.

• interaction (Interaction) –

  The interaction of the catch prompt, already deferred.

• player (Player) –

  The player trying to catch.

• guess (str) –

  The name typed by the user.

cancelled

cancelled: bool = field(default=False, init=False)

guess

guess: str

hook_name

hook_name = 'pre_catch'

interaction

interaction: Interaction

player

player: Player

reason

reason: str | None = field(default=None, init=False)

view

view: BallSpawnView

cancel

cancel(reason: str | None = None)

Block the action.

Parameters:

• reason (str | None, default: None ) –

  A message shown to the user. A generic message is shown if omitted.

Source code in ballsdex/core/hooks.py

def cancel(self, reason: str | None = None):
    """
    Block the action.

    Parameters
    ----------
    reason: str | None
        A message shown to the user. A generic message is shown if omitted.
    """
    self.cancelled = True
    self.reason = reason

PreGiveContext

PreGiveContext(*, sender: Player, recipient: Player)

Bases: CancellableContext

A player is about to give countryballs to another, or send them a donation request. Cancelling blocks the donation, and the reason is shown to the sender.

Attributes:

• sender (Player) –

  The player giving.

• recipient (Player) –

  The player receiving.

cancelled

cancelled: bool = field(default=False, init=False)

hook_name

hook_name = 'pre_give'

reason

reason: str | None = field(default=None, init=False)

recipient

recipient: Player

sender

sender: Player

cancel

cancel(reason: str | None = None)

Block the action.

Parameters:

• reason (str | None, default: None ) –

  A message shown to the user. A generic message is shown if omitted.

Source code in ballsdex/core/hooks.py

def cancel(self, reason: str | None = None):
    """
    Block the action.

    Parameters
    ----------
    reason: str | None
        A message shown to the user. A generic message is shown if omitted.
    """
    self.cancelled = True
    self.reason = reason

PreTradeContext

PreTradeContext(*, trade: TradeInstance, trader1: TradingUser, trader2: TradingUser)

Bases: CancellableContext

Both users confirmed a trade and it is about to be saved. Cancelling ends the trade without exchanging anything, and the reason is shown on the trade message.

Attributes:

• trade (TradeInstance) –

  The trade view.

• trader1 (TradingUser) –

  The user who started the trade.

• trader2 (TradingUser) –

  The other user of the trade.

cancelled

cancelled: bool = field(default=False, init=False)

hook_name

hook_name = 'pre_trade'

reason

reason: str | None = field(default=None, init=False)

trade

trade: TradeInstance

trader1

trader1: TradingUser

trader2

trader2: TradingUser

cancel

cancel(reason: str | None = None)

Block the action.

Parameters:

• reason (str | None, default: None ) –

  A message shown to the user. A generic message is shown if omitted.

Source code in ballsdex/core/hooks.py

def cancel(self, reason: str | None = None):
    """
    Block the action.

    Parameters
    ----------
    reason: str | None
        A message shown to the user. A generic message is shown if omitted.
    """
    self.cancelled = True
    self.reason = reason

SpawnCheckContext

SpawnCheckContext(*, guild: Guild, channel: TextChannel, message: Message, algo: str)

Bases: CancellableContext

A countryball is about to spawn naturally. Cancelling silently skips this spawn.

Attributes:

• guild (Guild) –

  The guild where the spawn happens.

• channel (TextChannel) –

  The configured spawn channel.

• message (Message) –

  The message that triggered the spawn.

• algo (str) –

  The spawn algorithm that decided to spawn.

algo

algo: str

cancelled

cancelled: bool = field(default=False, init=False)

channel

channel: TextChannel

guild

guild: Guild

hook_name

hook_name = 'spawn_check'

message

message: Message

reason

reason: str | None = field(default=None, init=False)

cancel

cancel(reason: str | None = None)

Block the action.

Parameters:

• reason (str | None, default: None ) –

  A message shown to the user. A generic message is shown if omitted.

Source code in ballsdex/core/hooks.py

def cancel(self, reason: str | None = None):
    """
    Block the action.

    Parameters
    ----------
    reason: str | None
        A message shown to the user. A generic message is shown if omitted.
    """
    self.cancelled = True
    self.reason = reason

hook

hook(name: str, *, priority: int = 0)

Mark a cog method as a hook. It is registered when the cog is added to the bot.

Parameters:

• name (str) –

  The name of the hook.

• priority (int, default: 0 ) –

  Hooks with a higher priority run first. Defaults to 0.

Source code in ballsdex/core/hooks.py

def hook(name: str, *, priority: int = 0):
    """
    Mark a cog method as a hook. It is registered when the cog is added to the bot.

    Parameters
    ----------
    name: str
        The name of the hook.
    priority: int
        Hooks with a higher priority run first. Defaults to 0.
    """
    if name not in HOOK_NAMES:
        raise ValueError(f"Unknown hook {name!r}")

    def decorator(func):
        func.__ballsdex_hook__ = (name, priority)
        return func

    return decorator