Skip to content

tux.cogs.moderation.action_mixin

Classes:

Name Description
ModerationActionMixin

Shared execution helper for moderation actions.

Classes

ModerationActionMixin(bot: Tux)

Bases: ModerationCogBase

Shared execution helper for moderation actions.

This mixin builds the actions list and delegates to :py:meth:ModerationCogBase.execute_mod_action so concrete commands only need to supply metadata and the Discord API coroutine.

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/action_mixin.py
Python
self,
ctx: commands.Context,  # noqa: D401
*,
member: MemberOrUser,
reason: str,
duration: str | None,
purge: int,
silent: bool,

Functions

_run_action(ctx: commands.Context, *, member: MemberOrUser, reason: str, duration: str | None, purge: int, silent: bool, case_type: CaseType, dm_verb: str, discord_action: Callable[[], Awaitable[Any]] | None = None) -> None async

Validate and execute the moderation workflow.

Source code in tux/cogs/moderation/action_mixin.py
Python
async def _run_action(  # noqa: WPS211  (many params is ok for clarity)
    self,
    ctx: commands.Context,  # noqa: D401
    *,
    member: MemberOrUser,
    reason: str,
    duration: str | None,
    purge: int,
    silent: bool,
    case_type: CaseType,
    dm_verb: str,
    discord_action: Callable[[], Awaitable[Any]] | None = None,
) -> None:
    """Validate and execute the moderation workflow."""

    assert ctx.guild is not None, "Guild-only command"  # noqa: S101

    # Permission / sanity checks
    if not await self.check_conditions(ctx, member, ctx.author, case_type.name.lower()):
        return

    if purge and not (0 <= purge <= 7):
        await ctx.send("Purge must be between 0 and 7 days.", ephemeral=True)
        return

    final_reason = reason or CONST.DEFAULT_REASON

    actions: list[tuple[Awaitable[Any], type[Any]]] = []
    if discord_action is not None:
        actions.append((discord_action(), type(None)))

    await self.execute_mod_action(
        ctx=ctx,
        case_type=case_type,
        user=member,
        reason=final_reason,
        silent=silent,
        dm_action=dm_verb,
        actions=actions,
        duration=duration,
    )
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/action_mixin.py
Python
    dm_verb: str,
    discord_action: Callable[[], Awaitable[Any]] | None = None,
) -> None:
    """Validate and execute the moderation workflow."""

    assert ctx.guild is not None, "Guild-only command"  # noqa: S101

    # Permission / sanity checks
    if not await self.check_conditions(ctx, member, ctx.author, case_type.name.lower()):
        return

    if purge and not (0 <= purge <= 7):
        await ctx.send("Purge must be between 0 and 7 days.", ephemeral=True)
        return

    final_reason = reason or CONST.DEFAULT_REASON

    actions: list[tuple[Awaitable[Any], type[Any]]] = []
    if discord_action is not None:
        actions.append((discord_action(), type(None)))

    await self.execute_mod_action(
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/action_mixin.py
Python
            user=member,
            reason=final_reason,
            silent=silent,
            dm_action=dm_verb,
            actions=actions,
            duration=duration,
        )


# ---------------------------------------------------------------------------
# No-op setup so the cog loader skips this util module
# ---------------------------------------------------------------------------


async def setup(bot):  # type: ignore[unused-argument]
    """Utility module – nothing to load."""
    return
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.
_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.

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
_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

setup(bot) async

Utility module – nothing to load.

Source code in tux/cogs/moderation/action_mixin.py
Python
async def setup(bot):  # type: ignore[unused-argument]
    """Utility module – nothing to load."""
    return