Registry

Registry is the shared base class for both Bot and Extension. It owns all command/group registration logic and event handler wiring, keeping that surface consistent across both entry points.

You typically interact with Registry indirectly through the decorators it exposes (@bot.command, @bot.on, etc.).

matrix.registry.Registry

Registry(name, prefix=None)

Base class providing shared registration behaviour for Bot and Extension.

Handles registration of commands, groups, events, checks, schedules, and error handlers. Subclasses must initialize the required attributes defined below, either directly or via super().__init__().

Source code in matrix/registry.py

def __init__(self, name: str, prefix: Optional[str] = None):
    self.name = name
    self.prefix = prefix
    self.log = logging.getLogger(__name__)

    self._commands: Dict[str, Command] = {}
    self._checks: List[Callback] = []
    self._scheduler: Scheduler = Scheduler()

    self._event_handlers: Dict[Type[Event], List[Callback]] = defaultdict(list)
    self._hook_handlers: Dict[str, List[Callback]] = defaultdict(list)
    self._error_handlers: Dict[type[Exception], ErrorCallback] = {}
    self._command_error_handlers: Dict[type[Exception], CommandErrorCallback] = {}

EVENT_MAP class-attribute instance-attribute

EVENT_MAP = {
    "on_typing": TypingNoticeEvent,
    "on_message": RoomMessageText,
    "on_react": ReactionEvent,
    "on_member_join": RoomMemberEvent,
    "on_member_leave": RoomMemberEvent,
    "on_member_invite": RoomMemberEvent,
    "on_member_ban": RoomMemberEvent,
    "on_member_kick": RoomMemberEvent,
    "on_member_change": RoomMemberEvent,
}

LIFECYCLE_EVENTS class-attribute instance-attribute

LIFECYCLE_EVENTS = {
    "on_ready",
    "on_error",
    "on_command",
    "on_command_error",
}

name instance-attribute

name = name

prefix instance-attribute

prefix = prefix

log instance-attribute

log = getLogger(__name__)

commands property

commands

command

command(
    name=None,
    *,
    description=None,
    usage=None,
    cooldown=None
)

Decorator to register a coroutine function as a command handler.

The command name defaults to the function name unless explicitly provided.

Example

@bot.command(description="Returns pong!")
async def ping(ctx):
    await ctx.reply("Pong!")

Source code in matrix/registry.py

def command(
    self,
    name: Optional[str] = None,
    *,
    description: Optional[str] = None,
    usage: Optional[str] = None,
    cooldown: Optional[tuple[int, float]] = None,
) -> Callable[[Callback], Command]:
    """Decorator to register a coroutine function as a command handler.

    The command name defaults to the function name unless
    explicitly provided.

    ## Example

    ```python
    @bot.command(description="Returns pong!")
    async def ping(ctx):
        await ctx.reply("Pong!")
    ```
    """

    def wrapper(func: Callback) -> Command:
        cmd = Command(
            func,
            name=name,
            description=description,
            prefix=self.prefix,
            usage=usage,
            cooldown=cooldown,
        )
        return self.register_command(cmd)

    return wrapper

register_command

register_command(cmd)

Register a Command instance directly.

Prefer the :meth:command decorator for typical use. This method is useful when constructing a Command object manually or when loading commands from an extension.

Source code in matrix/registry.py

def register_command(self, cmd: Command) -> Command:
    """Register a Command instance directly.

    Prefer the :meth:`command` decorator for typical use. This method
    is useful when constructing a ``Command`` object manually or when
    loading commands from an extension.
    """
    if cmd.name in self._commands:
        raise AlreadyRegisteredError(cmd)

    self._commands[cmd.name] = cmd
    logger.debug("command '%s' registered on %s", cmd, type(self).__name__)

    return cmd

group

group(
    name=None,
    *,
    description=None,
    usage=None,
    cooldown=None
)

Decorator to register a coroutine function as a command group.

A group acts as a parent command that can have subcommands attached to it via its own @group.command() decorator. The group name defaults to the function name unless explicitly provided.

Example

@bot.group(description="Group of mathematical commands")
async def math(ctx):
    await ctx.reply("You called !math")

@math.command()
async def add(ctx, a: int, b: int):
    await ctx.reply(f"{a} + {b} = {a + b}")

@math.command()
async def subtract(ctx, a: int, b: int):
    await ctx.reply(f"{a} - {b} = {a - b}")

Source code in matrix/registry.py

def group(
    self,
    name: Optional[str] = None,
    *,
    description: Optional[str] = None,
    usage: Optional[str] = None,
    cooldown: Optional[tuple[int, float]] = None,
) -> Callable[[Callback], Group]:
    """Decorator to register a coroutine function as a command group.

    A group acts as a parent command that can have subcommands attached
    to it via its own ``@group.command()`` decorator. The group name
    defaults to the function name unless explicitly provided.

    ## Example

    ```python
    @bot.group(description="Group of mathematical commands")
    async def math(ctx):
        await ctx.reply("You called !math")

    @math.command()
    async def add(ctx, a: int, b: int):
        await ctx.reply(f"{a} + {b} = {a + b}")

    @math.command()
    async def subtract(ctx, a: int, b: int):
        await ctx.reply(f"{a} - {b} = {a - b}")
    ```
    """

    def wrapper(func: Callback) -> Group:
        grp = Group(
            func,
            name=name,
            description=description,
            prefix=self.prefix,
            usage=usage,
            cooldown=cooldown,
        )
        return self.register_group(grp)

    return wrapper

register_group

register_group(group)

Register a Group instance directly.

Prefer the :meth:group decorator for typical use. This method is useful when constructing a Group object manually or when loading groups from an extension.

Source code in matrix/registry.py

def register_group(self, group: Group) -> Group:
    """Register a Group instance directly.

    Prefer the :meth:`group` decorator for typical use. This method
    is useful when constructing a ``Group`` object manually or when
    loading groups from an extension.
    """
    if group.name in self._commands:
        raise AlreadyRegisteredError(group)

    self._commands[group.name] = group
    logger.debug("group '%s' registered on %s", group, type(self).__name__)

    return group

event

event(func=None, *, event_spec=None)

Decorator to register a coroutine as an event handler.

Can be used with or without arguments. Without arguments, the event type is inferred from the function name via EVENT_MAP. Multiple handlers for the same event type are supported and called in registration order.

Example

@bot.event
async def on_message(room, event):
    ...

@bot.event(event_spec=RoomMemberEvent)
async def handle_member(room, event):
    ...

@bot.event(event_spec="on_member_join")
async def welcome(room, event):
    ...

Source code in matrix/registry.py

def event(
    self,
    func: Optional[Callback] = None,
    *,
    event_spec: Union[str, Type[Event], None] = None,
) -> Union[Callback, Callable[[Callback], Callback]]:
    """Decorator to register a coroutine as an event handler.

    Can be used with or without arguments. Without arguments, the event
    type is inferred from the function name via ``EVENT_MAP``. Multiple
    handlers for the same event type are supported and called in
    registration order.

    ## Example

    ```python
    @bot.event
    async def on_message(room, event):
        ...

    @bot.event(event_spec=RoomMemberEvent)
    async def handle_member(room, event):
        ...

    @bot.event(event_spec="on_member_join")
    async def welcome(room, event):
        ...
    ```
    """

    def wrapper(f: Callback) -> Callback:
        if not inspect.iscoroutinefunction(f):
            raise TypeError("Event handlers must be coroutines")

        key = event_spec if isinstance(event_spec, str) else f.__name__
        event_type: type[Event] | None = (
            event_spec
            if event_spec and not isinstance(event_spec, str)
            else self.EVENT_MAP.get(key)
        )

        if event_type is None:
            raise ValueError(f"Unknown event: {key!r}")

        return self.register_event(event_type, f)

    if func is None:
        return wrapper
    return wrapper(func)

register_event

register_event(event_type, callback)

Register an event handler directly for a given event type.

Prefer the :meth:event decorator for typical use. This method is useful when loading event handlers from an extension.

Source code in matrix/registry.py

def register_event(self, event_type: Type[Event], callback: Callback) -> Callback:
    """Register an event handler directly for a given event type.

    Prefer the :meth:`event` decorator for typical use. This method
    is useful when loading event handlers from an extension.
    """
    self._event_handlers[event_type].append(callback)
    logger.debug(
        "registered event %s for %s", callback.__name__, event_type.__name__
    )
    return callback

hook

hook(func=None, *, event_name=None)

Decorator to register a coroutine as a lifecycle event hook.

Lifecycle events include things like on_ready, on_command, and on_error. If the event name is not provided, it is inferred from the function name. Multiple handlers for the same lifecycle event are supported and called in registration order.

Example

@bot.hook
async def on_ready():
    print("Bot is ready!")

@bot.hook(event_name="on_command")
async def log_command(ctx):
    print(f"Command invoked: {ctx.command}")

Source code in matrix/registry.py

def hook(
    self, func: Optional[Callback] = None, *, event_name: Optional[str] = None
) -> Union[Callback, Callable[[Callback], Callback]]:
    """Decorator to register a coroutine as a lifecycle event hook.

    Lifecycle events include things like ``on_ready``, ``on_command``,
    and ``on_error``. If the event name is not provided, it is inferred
    from the function name. Multiple handlers for the same lifecycle
    event are supported and called in registration order.

    ## Example

    ```python
    @bot.hook
    async def on_ready():
        print("Bot is ready!")

    @bot.hook(event_name="on_command")
    async def log_command(ctx):
        print(f"Command invoked: {ctx.command}")
    ```
    """

    def wrapper(f: Callback) -> Callback:
        if not inspect.iscoroutinefunction(f):
            raise TypeError("Lifecycle hooks must be coroutines")

        name = event_name or f.__name__
        if name not in self.LIFECYCLE_EVENTS:
            raise ValueError(f"Unknown lifecycle event: {name}")

        return self.register_hook(name, f)

    if func is None:
        return wrapper
    return wrapper(func)

register_hook

register_hook(event_name, callback)

Register a lifecycle event hook directly for a given event name.

Prefer the :meth:hook decorator for typical use. This method is useful when loading lifecycle hooks from an extension.

Source code in matrix/registry.py

def register_hook(self, event_name: str, callback: Callback) -> Callback:
    """Register a lifecycle event hook directly for a given event name.

    Prefer the :meth:`hook` decorator for typical use. This method
    is useful when loading lifecycle hooks from an extension.
    """
    if not inspect.iscoroutinefunction(callback):
        raise TypeError("Lifecycle hooks must be coroutines")

    if event_name not in self.LIFECYCLE_EVENTS:
        raise ValueError(f"Unknown lifecycle event: {event_name}")

    self._hook_handlers[event_name].append(callback)
    logger.debug(
        "registered lifecycle hook '%s' for event '%s' on %s",
        callback.__name__,
        event_name,
        type(self).__name__,
    )
    return callback

check

check(func)

Register a global check that must pass before any command is invoked.

The check receives the current :class:Context and must return a boolean. If any check returns False, a :class:CheckError is raised and the command is not executed.

Example

@bot.check
async def is_not_banned(ctx):
    return ctx.sender not in banned_users

Source code in matrix/registry.py

def check(self, func: Callback) -> Callback:
    """Register a global check that must pass before any command is invoked.

    The check receives the current :class:`Context` and must return a
    boolean. If any check returns ``False``, a :class:`CheckError` is
    raised and the command is not executed.

    ## Example

    ```python
    @bot.check
    async def is_not_banned(ctx):
        return ctx.sender not in banned_users
    ```
    """
    if not inspect.iscoroutinefunction(func):
        raise TypeError("Checks must be coroutines")

    self._checks.append(func)
    logger.debug("registered check '%s' on %s", func.__name__, type(self).__name__)

    return func

schedule

schedule(cron)

Decorator to register a coroutine as a scheduled task.

When used on an extension, scheduled tasks are merged into the bot's scheduler when the extension is loaded.

Example

@bot.schedule("0 9 * * *")
async def morning_message():
    await room.send("Good morning!")

Source code in matrix/registry.py

def schedule(self, cron: str) -> Callable[[Callback], Callback]:
    """Decorator to register a coroutine as a scheduled task.

    When used on an extension, scheduled tasks are merged into the
    bot's scheduler when the extension is loaded.

    ## Example

    ```python
    @bot.schedule("0 9 * * *")
    async def morning_message():
        await room.send("Good morning!")
    ```
    """

    def wrapper(f: Callback) -> Callback:
        if not inspect.iscoroutinefunction(f):
            raise TypeError("Scheduled tasks must be coroutines")

        self._scheduler.schedule(cron, f)
        logger.debug(
            "scheduled '%s' for cron '%s' on %s",
            f.__name__,
            cron,
            type(self).__name__,
        )

        return f

    return wrapper

error

error(
    exception: Optional[type[Exception]] = None,
    *,
    context: Literal[True]
) -> Callable[[CommandErrorCallback], CommandErrorCallback]

error(
    exception: Optional[type[Exception]] = None,
    *,
    context: Literal[False] = ...
) -> Callable[[ErrorCallback], ErrorCallback]

error(exception=None, *, context=False)

Decorator to register an error handler.

If an exception type is provided, the handler is only invoked for that specific exception. If omitted, the handler acts as a generic fallback for any unhandled error.

Set context=True to receive the command context alongside the error, useful for command-specific errors where you want to reply to the user.

Example

@bot.error(ValueError)
async def on_value_error(error):
    pass

@bot.error()
async def on_any_error(error):
    pass

@bot.error(CommandNotFoundError, context=True)
async def on_command_not_found(ctx, error):
    await ctx.reply("Command not found!")

Source code in matrix/registry.py

def error(
    self,
    exception: Optional[type[Exception]] = None,
    *,
    context: bool = False,
) -> Union[
    Callable[[ErrorCallback], ErrorCallback],
    Callable[[CommandErrorCallback], CommandErrorCallback],
]:
    """Decorator to register an error handler.

    If an exception type is provided, the handler is only invoked for
    that specific exception. If omitted, the handler acts as a generic
    fallback for any unhandled error.

    Set ``context=True`` to receive the command context alongside the error,
    useful for command-specific errors where you want to reply to the user.

    ## Example

    ```python
    @bot.error(ValueError)
    async def on_value_error(error):
        pass

    @bot.error()
    async def on_any_error(error):
        pass

    @bot.error(CommandNotFoundError, context=True)
    async def on_command_not_found(ctx, error):
        await ctx.reply("Command not found!")
    ```
    """

    def wrapper(
        func: F,
    ) -> F:
        if not inspect.iscoroutinefunction(func):
            raise TypeError("Error handlers must be coroutines")

        if context:
            self._register_command_error(
                cast(CommandErrorCallback, func), exception
            )
        else:
            self._register_error(cast(ErrorCallback, func), exception)

        logger.debug(
            "registered error handler '%s' on %s",
            func.__name__,
            type(self).__name__,
        )
        return func

    return wrapper