Skip to content

tux.cogs.moderation.command_meta

Metaclass + base class for declarative moderation commands.

Subclass ModerationCommand once per moderation action. Example::

Text Only
class Ban(ModerationCommand):
    name = "ban"
    aliases = ["b"]
    case_type = CaseType.BAN
    required_pl = 3
    flags = {
        "purge": dict(type=int, aliases=["p"], default=0, desc="Days to delete"),
        "silent": dict(type=bool, aliases=["s", "quiet"], default=False, desc="No DM"),
    }

    async def _action(self, guild: discord.Guild, member: discord.Member, *, flags, reason):
        await guild.ban(member, reason=reason, delete_message_seconds=flags.purge * 86400)

Classes:

Name Description
ModerationCommandMeta

Metaclass that turns class attributes into a real command.

ModerationCommand

Base class to inherit for each moderation action.

ModerationCommandsCog

Classes

ModerationCommandMeta

Bases: type

Metaclass that turns class attributes into a real command.

ModerationCommand

Base class to inherit for each moderation action.

ModerationCommandsCog(bot: commands.Bot)

Bases: ModerationCogBase

Methods:

Name Description
get_user_lock

Get or create a lock for operations on a specific user.

clean_user_locks

Remove locks for users that are not currently in use.

execute_user_action_with_lock

Execute an action on a user with a lock to prevent race conditions.

execute_mod_action

Execute a moderation action with case creation, DM sending, and additional actions.

send_error_response

Send a standardized error response.

create_embed

Create an embed for moderation actions.

send_embed

Send an embed to the log channel.

send_dm

Send a DM to the target user.

check_conditions

Check if the conditions for the moderation action are met.

handle_case_response

Handle the response for a case.

is_pollbanned

Check if a user is poll banned.

is_snippetbanned

Check if a user is snippet banned.

is_jailed

Check if a user is jailed using the optimized latest case method.

execute_mixed_mod_action

Parse mixed_args according to config and execute the moderation flow.

execute_flag_mod_action

Execute moderation flow based on flags parsed by FlagConverter.

Source code in tux/cogs/moderation/command_meta.py
Python
def __init__(self, bot: commands.Bot):
    super().__init__(bot)  # type: ignore[arg-type]

    for cls in _REGISTRY:
        self.bot.add_command(cls.text_command)  # type: ignore[attr-defined]
        self.bot.tree.add_command(cls.slash_command)  # type: ignore[attr-defined]

Functions

get_user_lock(user_id: int) -> Lock async

Get or create a lock for operations on a specific user. If the number of stored locks exceeds the cleanup threshold, unused locks are removed.

Parameters:

Name Type Description Default
user_id int

The ID of the user to get a lock for.

required

Returns:

Type Description
Lock

The lock for the user.

Source code in tux/cogs/moderation/command_meta.py
Python
def __new__(mcls, name: str, bases: tuple[type, ...], ns: dict[str, Any]):
    cls = super().__new__(mcls, name, bases, ns)
    if cls.__name__ == "ModerationCommand":
        return cls

    # Extract required attributes
    cmd_name: str = getattr(cls, "name")  # type: ignore[arg-type]
    aliases: list[str] = getattr(cls, "aliases", [])  # type: ignore[arg-type]
    case_type = getattr(cls, "case_type")
    required_pl: int = getattr(cls, "required_pl", 0)
    flags_spec: Dict[str, Dict[str, Any]] = getattr(cls, "flags", {})  # type: ignore[arg-type]
    description: str = getattr(cls, "description", cmd_name.title())

    # Build FlagConverter
    FlagsCls = build_flag_converter(
        cmd_name,
        duration="duration" in flags_spec,
        purge="purge" in flags_spec,
        silent="silent" in flags_spec,
    )

    # Make FlagsCls resolvable for annotation eval
clean_user_locks() -> None async

Remove locks for users that are not currently in use. Iterates through the locks and removes any that are not currently locked.

Source code in tux/cogs/moderation/command_meta.py
Python
    # Expose under a stable alias so eval("FlagsCls") always succeeds
    globals()['FlagsCls'] = FlagsCls
    setattr(builtins, 'FlagsCls', FlagsCls)

# --------------------------------------------------
# Shared executor
# --------------------------------------------------
async def _run(self: ModerationCogBase, ctx, target: MemberOrUser, flags, reason: str):
    if not await self.check_conditions(ctx, target, ctx.author, cmd_name):
        return

    silent = getattr(flags, "silent", False)
    duration = getattr(flags, "duration", None)

    action_coro = cls._action(self, ctx.guild, target, flags=flags, reason=reason)  # type: ignore[arg-type]
    actions = [(action_coro, type(None))]
execute_user_action_with_lock(user_id: int, action_func: Callable[..., Coroutine[Any, Any, R]], *args: Any, **kwargs: Any) -> R async

Execute an action on a user with a lock to prevent race conditions.

Parameters:

Name Type Description Default
user_id int

The ID of the user to lock.

required
action_func Callable[..., Coroutine[Any, Any, R]]

The coroutine function to execute.

required
*args Any

Arguments to pass to the function.

()
**kwargs Any

Keyword arguments to pass to the function.

{}

Returns:

Type Description
R

The result of the action function.

Source code in tux/cogs/moderation/command_meta.py
Python
        ctx=ctx,
        case_type=case_type,
        user=target,
        reason=reason,
        silent=silent,
        dm_action=getattr(cls, "dm_action", cmd_name),
        actions=actions,
        duration=duration,
    )

# --------------------------------------------------
# Text command (prefix)
# --------------------------------------------------
async def _text(self: ModerationCogBase, ctx: commands.Context, target: MemberOrUser, *, flags: FlagsCls | None = None, reason: str = "") -> None:  # type: ignore[arg-type]
    if flags is None:
        flags = FlagsCls()  # type: ignore[assignment]
    await _run(self, ctx, target, flags, reason)

if FlagsCls is not None:
    _text.__globals__[FlagsCls.__name__] = FlagsCls
    _text.__globals__['FlagsCls'] = FlagsCls  # also as generic alias
    from typing import Dict as _Dict  # noqa: WPS433
    _text.__globals__.setdefault('Dict', _Dict)

_text.__name__ = cmd_name
_text.__doc__ = description

text_cmd = commands.command(name=cmd_name, aliases=aliases, help=description)(_text)
# Override usage string to exclude internal ctx parameter
text_cmd.usage = f"{cmd_name} <target> <flags> <reason>"
_dummy_action() -> None async

Dummy coroutine for moderation actions that only create a case without performing Discord API actions. Used by commands like warn, pollban, snippetban etc. that only need case creation.

Source code in tux/cogs/moderation/command_meta.py
Python
# --------------------------------------------------
# Slash command
# --------------------------------------------------
async def _slash(
    interaction: discord.Interaction,
    target: MemberOrUser,
execute_mod_action(ctx: commands.Context[Tux], case_type: CaseType, user: discord.Member | discord.User, reason: str, silent: bool, dm_action: str, actions: Sequence[tuple[Any, type[R]]] = (), duration: str | None = None, expires_at: datetime | None = None) -> None async

Execute a moderation action with case creation, DM sending, and additional actions.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
case_type CaseType

The type of case to create.

required
user Union[Member, User]

The target user of the moderation action.

required
reason str

The reason for the moderation action.

required
silent bool

Whether to send a DM to the user.

required
dm_action str

The action description for the DM.

required
actions Sequence[tuple[Any, type[R]]]

Additional actions to execute and their expected return types.

()
duration Optional[str]

The duration of the action, if applicable (for display/logging).

None
expires_at Optional[datetime]

The specific expiration time, if applicable.

None
Source code in tux/cogs/moderation/command_meta.py
Python
            duration: str | None = None,
            purge: int = 0,
            silent: bool = False,
            reason: str = "",
        ) -> None:  # type: ignore[arg-type]
            """App command callback (no bound self) using explicit options."""

            from types import SimpleNamespace  # noqa: WPS433

            flags_obj = SimpleNamespace(duration=duration, purge=purge, silent=silent)

            bot = interaction.client  # type: ignore[attr-defined]
            cog: ModerationCogBase | None = bot.get_cog("ModerationCommandsCog")  # type: ignore[attr-defined]
            if cog is None:
                return

            ctx = await cog.bot.get_context(interaction)  # type: ignore[attr-defined]
            await _run(cog, ctx, target, flags_obj, reason)

        if FlagsCls is not None:
            _slash.__globals__[FlagsCls.__name__] = FlagsCls
            _slash.__globals__['FlagsCls'] = FlagsCls

        slash_cmd = discord.app_commands.command(name=cmd_name, description=description)(_slash)

        # store on cls
        cls.text_command = text_cmd  # type: ignore[attr-defined]
        cls.slash_command = slash_cmd  # type: ignore[attr-defined]

        # register class
        _REGISTRY.append(cls)
        return cls


class ModerationCommand(metaclass=ModerationCommandMeta):
    """Base class to inherit for each moderation action."""

    name: ClassVar[str]
    aliases: ClassVar[list[str]] = []
    case_type: ClassVar[Any]
    required_pl: ClassVar[int] = 0
    flags: ClassVar[Dict[str, Dict[str, Any]]] = {}
    description: ClassVar[str] = ""

    # Child classes must implement _action
    async def _action(self, guild: discord.Guild, member: discord.Member | discord.User, *, flags: Any, reason: str) -> None:  # noqa: D401
        raise NotImplementedError


# Cog that loads all ModerationCommand subclasses
class ModerationCommandsCog(ModerationCogBase):
    def __init__(self, bot: commands.Bot):
        super().__init__(bot)  # type: ignore[arg-type]

        for cls in _REGISTRY:
            self.bot.add_command(cls.text_command)  # type: ignore[attr-defined]
            self.bot.tree.add_command(cls.slash_command)  # type: ignore[attr-defined]


async def setup(bot: commands.Bot):
    # Ensure all command modules are imported so subclasses register
    import importlib
    importlib.import_module("tux.cogs.moderation.commands")

    await bot.add_cog(ModerationCommandsCog(bot))
_handle_dm_result(user: discord.Member | discord.User, dm_result: Any) -> bool

Handle the result of sending a DM.

Parameters:

Name Type Description Default
user Union[Member, User]

The user the DM was sent to.

required
dm_result Any

The result of the DM sending operation.

required

Returns:

Type Description
bool

Whether the DM was successfully sent.

send_error_response(ctx: commands.Context[Tux], error_message: str, error_detail: Exception | None = None, ephemeral: bool = True) -> None async

Send a standardized error response.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
error_message str

The error message to display.

required
error_detail Optional[Exception]

The exception details, if available.

None
ephemeral bool

Whether the message should be ephemeral.

True
create_embed(ctx: commands.Context[Tux], title: str, fields: list[tuple[str, str, bool]], color: int, icon_url: str, timestamp: datetime | None = None, thumbnail_url: str | None = None) -> discord.Embed

Create an embed for moderation actions.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
title str

The title of the embed.

required
fields list[tuple[str, str, bool]]

The fields to add to the embed.

required
color int

The color of the embed.

required
icon_url str

The icon URL for the embed.

required
timestamp Optional[datetime]

The timestamp for the embed.

None
thumbnail_url Optional[str]

The thumbnail URL for the embed.

None

Returns:

Type Description
Embed

The embed for the moderation action.

send_embed(ctx: commands.Context[Tux], embed: discord.Embed, log_type: str) -> None async

Send an embed to the log channel.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
embed Embed

The embed to send.

required
log_type str

The type of log to send the embed to.

required
send_dm(ctx: commands.Context[Tux], silent: bool, user: discord.Member | discord.User, reason: str, action: str) -> bool async

Send a DM to the target user.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
silent bool

Whether the command is silent.

required
user Union[Member, User]

The target of the moderation action.

required
reason str

The reason for the moderation action.

required
action str

The action being performed.

required

Returns:

Type Description
bool

Whether the DM was successfully sent.

check_conditions(ctx: commands.Context[Tux], user: discord.Member | discord.User, moderator: discord.Member | discord.User, action: str) -> bool async

Check if the conditions for the moderation action are met.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
user Union[Member, User]

The target of the moderation action.

required
moderator Union[Member, User]

The moderator of the moderation action.

required
action str

The action being performed.

required

Returns:

Type Description
bool

Whether the conditions are met.

handle_case_response(ctx: commands.Context[Tux], case_type: CaseType, case_number: int | None, reason: str, user: discord.Member | discord.User, dm_sent: bool, duration: str | None = None) -> None async

Handle the response for a case.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
case_type CaseType

The type of case.

required
case_number Optional[int]

The case number.

required
reason str

The reason for the case.

required
user Union[Member, User]

The target of the case.

required
dm_sent bool

Whether the DM was sent.

required
duration Optional[str]

The duration of the case.

None
_format_case_title(case_type: CaseType, case_number: int | None, duration: str | None) -> str

Format a case title.

Parameters:

Name Type Description Default
case_type CaseType

The type of case.

required
case_number Optional[int]

The case number.

required
duration Optional[str]

The duration of the case.

required

Returns:

Type Description
str

The formatted case title.

is_pollbanned(guild_id: int, user_id: int) -> bool async

Check if a user is poll banned.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is poll banned, False otherwise.

is_snippetbanned(guild_id: int, user_id: int) -> bool async

Check if a user is snippet banned.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is snippet banned, False otherwise.

is_jailed(guild_id: int, user_id: int) -> bool async

Check if a user is jailed using the optimized latest case method.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is jailed, False otherwise.

execute_mixed_mod_action(ctx: commands.Context[Tux], config: ModerationCommandConfig, user: discord.Member | discord.User, mixed_args: str) -> None async

Parse mixed_args according to config and execute the moderation flow.

This serves as the single entry-point for all dynamically generated moderation commands. It handles: 1. Mixed-argument parsing (positional + flags). 2. Validation based on config (duration required?, purge range?, etc.). 3. Permission / sanity checks via check_conditions. 4. Building the actions list and delegating to :py:meth:execute_mod_action.

_validate_args(config: ModerationCommandConfig, parsed: dict[str, Any]) -> tuple[bool, dict[str, Any]]

Validate parsed arguments against config rules.

Returns (is_valid, validated_dict). On failure sends the error message via the ctx stored in validated_dict["ctx"] and returns False.

execute_flag_mod_action(ctx: commands.Context[Tux], config: ModerationCommandConfig, user: discord.Member | discord.User, flags: Any, reason: str) -> None async

Execute moderation flow based on flags parsed by FlagConverter.

This is the preferred pathway for dynamically generated moderation commands that rely on discord.py's native FlagConverter parsing.

Functions