Skip to content

moderation

Moderation services for Tux Bot such as case service, communication service and execution service.

Modules:

Classes:

Classes

CaseService

Python
CaseService(case_controller: CaseController)

Service for managing moderation cases.

Provides clean, testable methods for case operations without the complexity of mixin inheritance.

Initialize the case service.

Parameters:

  • case_controller (CaseController) –

    Database controller for case operations.

Used by:

Methods:

Functions

create_case async
Python
create_case(
    guild_id: int,
    user_id: int,
    moderator_id: int,
    case_type: CaseType,
    reason: str,
    **kwargs: Any,
) -> Case

Create a new moderation case.

Parameters:

  • guild_id (int) –

    ID of the guild.

  • user_id (int) –

    ID of the target user.

  • moderator_id (int) –

    ID of the moderator.

  • case_type (CaseType) –

    Type of moderation action.

  • reason (str) –

    Reason for the action.

  • **kwargs (Any, default: {} ) –

    Additional case data (use case_expires_at for expiration datetime).

Returns:

  • Case –

    The created case.

get_case async
Python
get_case(case_id: int) -> Case | None

Get a case by ID.

Parameters:

  • case_id (int) –

    The case ID to retrieve.

Returns:

  • Case | None –

    The case if found, None otherwise.

get_user_cases async
Python
get_user_cases(user_id: int, guild_id: int) -> list[Case]

Get all cases for a user in a guild.

Parameters:

  • user_id (int) –

    The user ID.

  • guild_id (int) –

    The guild ID.

Returns:

  • list[Case] –

    List of cases for the user.

get_active_cases async
Python
get_active_cases(user_id: int, guild_id: int) -> list[Case]

Get active cases for a user in a guild.

Parameters:

  • user_id (int) –

    The user ID.

  • guild_id (int) –

    The guild ID.

Returns:

  • list[Case] –

    List of active cases for the user.

update_mod_log_message_id async
Python
update_mod_log_message_id(case_id: int, message_id: int) -> Case | None

Update the mod log message ID for a case.

Parameters:

  • case_id (int) –

    The case ID to update.

  • message_id (int) –

    The Discord message ID from the mod log.

Returns:

  • Case | None –

    The updated case, or None if not found.

CommunicationService

Python
CommunicationService(bot: Tux)

Service for handling moderation-related communication.

Manages DM sending, embed creation, and user notifications with proper error handling and timeouts.

Initialize the communication service.

Parameters:

  • bot (Tux) –

    The Discord bot instance.

Used by:

Methods:

Functions

send_dm async
Python
send_dm(
    ctx: Context[Tux], silent: bool, user: Member | User, reason: str, dm_action: str
) -> bool

Send a DM to a user about a moderation action.

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • silent (bool) –

    Whether to send DM (if False, returns False immediately).

  • user (Member | User) –

    Target user.

  • reason (str) –

    Reason for the action.

  • dm_action (str) –

    Action description for DM.

Returns:

  • bool –

    True if DM was sent successfully, False otherwise.

send_error_response async
Python
send_error_response(
    ctx: Context[Tux] | Interaction, message: str, ephemeral: bool = True
) -> None

Send an error response to the user.

Parameters:

  • ctx (Context[Tux] | Interaction) –

    Command context.

  • message (str) –

    Error message to send.

  • ephemeral (bool, default: True ) –

    Whether the response should be ephemeral, by default True.

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

Create a moderation embed.

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • title (str) –

    Embed title.

  • fields (list[tuple[str, str, bool]]) –

    List of (name, value, inline) tuples.

  • color (int) –

    Embed color.

  • icon_url (str) –

    Icon URL for the embed.

  • timestamp (datetime | None, default: None ) –

    Optional timestamp, by default None.

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

    Optional thumbnail URL, by default None.

Returns:

  • Embed –

    The created embed.

send_embed async
Python
send_embed(ctx: Context[Tux], embed: Embed, log_type: str = 'mod') -> Message | None

Send an embed and optionally log it.

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • embed (Embed) –

    The embed to send.

  • log_type (str, default: 'mod' ) –

    Type of log entry, by default "mod".

Returns:

  • Message | None –

    The sent message if successful.

send_audit_log_embed async
Python
send_audit_log_embed(ctx: Context[Tux], embed: Embed) -> Message | None

Send an embed to the audit log channel.

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • embed (Embed) –

    The embed to send to audit log.

Returns:

  • Message | None –

    The sent audit log message if successful, None otherwise.

send_mod_log_embed async
Python
send_mod_log_embed(ctx: Context[Tux], embed: Embed) -> Message | None

Send an embed to the mod log channel.

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • embed (Embed) –

    The embed to send to mod log.

Returns:

  • Message | None –

    The sent mod log message if successful, None otherwise.

_create_dm_embed
Python
_create_dm_embed(action: str, reason: str, moderator: User) -> Embed

Create a DM embed for moderation actions.

Parameters:

  • action (str) –

    The action that was taken.

  • reason (str) –

    Reason for the action.

  • moderator (User) –

    The moderator who performed the action.

Returns:

  • Embed –

    The DM embed.

ExecutionService

Python
ExecutionService(
    failure_threshold: int = 5,
    recovery_timeout: float = 60.0,
    max_retries: int = 3,
    base_delay: float = 1.0,
)

Service for executing moderation actions with retry logic.

Provides circuit breaker patterns and proper error handling for Discord API operations.

Initialize the execution service.

Parameters:

  • failure_threshold (int, default: 5 ) –

    Number of failures before opening circuit breaker, by default 5.

  • recovery_timeout (float, default: 60.0 ) –

    Seconds to wait before retrying after circuit opens, by default 60.0.

  • max_retries (int, default: 3 ) –

    Maximum number of retry attempts for operations, by default 3.

  • base_delay (float, default: 1.0 ) –

    Base delay in seconds for exponential backoff, by default 1.0.

Used by:

Methods:

Functions

execute_with_retry async
Python
execute_with_retry(
    operation_type: str,
    action: Callable[..., Coroutine[Any, Any, Any]],
    *args: Any,
    **kwargs: Any,
) -> Any

Execute an action with retry logic and circuit breaker.

Parameters:

  • operation_type (str) –

    Type of operation for circuit breaker.

  • action (Callable[..., Coroutine[Any, Any, Any]]) –

    The async callable to execute (must be a callable, not a coroutine).

  • *args (Any, default: () ) –

    Positional arguments for the action.

  • **kwargs (Any, default: {} ) –

    Keyword arguments for the action.

Returns:

  • Any –

    The result of the action.

Raises:

  • RuntimeError –

    If the circuit breaker is open for this operation type.

  • Forbidden –

    If the bot lacks permissions.

  • HTTPException –

    If a Discord API error occurs.

  • NotFound –

    If the resource is not found.

_is_circuit_open
Python
_is_circuit_open(operation_type: str) -> bool

Check if the circuit breaker is open for an operation type.

Parameters:

  • operation_type (str) –

    The operation type to check.

Returns:

  • bool –

    True if circuit is open, False otherwise.

_record_success
Python
_record_success(operation_type: str) -> None

Record a successful operation.

Parameters:

  • operation_type (str) –

    The operation type.

_record_failure
Python
_record_failure(operation_type: str) -> None

Record a failed operation.

Parameters:

  • operation_type (str) –

    The operation type.

_calculate_delay
Python
_calculate_delay(attempt: int, base_delay: float) -> float

Calculate delay for retry with exponential backoff.

Parameters:

  • attempt (int) –

    The current attempt number (0-based).

  • base_delay (float) –

    Base delay in seconds.

Returns:

  • float –

    Delay in seconds.

get_operation_type
Python
get_operation_type(case_type: CaseType) -> str

Get the operation type for circuit breaker based on case type.

Uses the case type name directly as the operation type for simplicity and clear correlation between operations and their failure patterns.

Parameters:

  • case_type (CaseType) –

    The case type.

Returns:

  • str –

    Operation type string for circuit breaker configuration.

ModerationServiceFactory

Factory for creating moderation service instances.

Centralizes the creation logic for moderation services to ensure consistent dependency injection across all moderation cogs.

Methods:

Functions

create_coordinator staticmethod
Python
create_coordinator(bot: Tux, case_controller: CaseController) -> ModerationCoordinator

Create a ModerationCoordinator with all required services.

Parameters:

  • bot (Tux) –

    The bot instance for communication service

  • case_controller (CaseController) –

    The database controller for case management

Returns:

Examples:

Python Console Session
>>> coordinator = ModerationServiceFactory.create_coordinator(self.bot, self.db.case)

ModerationCoordinator

Python
ModerationCoordinator(
    case_service: CaseService,
    communication_service: CommunicationService,
    execution_service: ExecutionService,
)

Main coordinator for moderation operations.

Orchestrates case creation, communication, and execution using proper service composition instead of mixins.

Initialize the moderation coordinator.

Parameters:

Returned by:

Methods:

Functions

execute_moderation_action async
Python
execute_moderation_action(
    ctx: Context[Tux],
    case_type: CaseType,
    user: Member | User,
    reason: str,
    silent: bool = False,
    dm_action: str | None = None,
    actions: Sequence[tuple[Callable[..., Coroutine[Any, Any, Any]], type[Any]]]
    | None = None,
    duration: int | None = None,
    expires_at: datetime | None = None,
    **extra_case_data: Any,
) -> Case | None

Execute a complete moderation action.

This method orchestrates the entire moderation flow: 1. Validate permissions and inputs 2. Send DM if required (before action for removal actions) 3. Execute Discord actions with retry logic 4. Create database case 5. Send DM if required (after action for non-removal actions) 6. Send response embed to the moderator 7. Send response embed to the log channel 8. Update the case audit log message ID

Parameters:

  • ctx (Context[Tux]) –

    Command context.

  • case_type (CaseType) –

    Type of moderation action.

  • user (Member | User) –

    Target user.

  • reason (str) –

    Reason for the action.

  • silent (bool, default: False ) –

    Whether to send DM to user, by default False.

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

    Custom DM action description, by default None.

  • actions (Sequence[tuple[Callable[..., Coroutine[Any, Any, Any]], type[Any]]] | None, default: None ) –

    Discord API actions to execute, by default None.

  • duration (int | None, default: None ) –

    Duration for temp actions, by default None.

  • expires_at (datetime | None, default: None ) –

    Expiration timestamp for temp actions, by default None.

  • **extra_case_data (Any, default: {} ) –

    Additional case data fields.

Returns:

  • Case | None –

    The created case, or None if case creation failed.

_handle_dm_timing async
Python
_handle_dm_timing(
    ctx: Context[Tux],
    case_type: CaseType,
    user: Member | User,
    reason: str,
    action_desc: str,
    silent: bool,
) -> bool

Handle DM timing based on action type.

Returns:

  • True if DM was sent, False otherwise –
_execute_actions async
Python
_execute_actions(
    ctx: Context[Tux],
    case_type: CaseType,
    user: Member | User,
    actions: Sequence[tuple[Callable[..., Coroutine[Any, Any, Any]], type[Any]]],
) -> list[Any]

Execute Discord API actions.

Note: Error handling is now centralized in the error handler. Exceptions are allowed to bubble up to be properly handled by the centralized error handler, which provides: - Consistent error messaging - Proper Sentry integration with command context - Guild/user context enrichment - Transaction management

Returns:

  • List of action results –
_handle_post_action_dm async
Python
_handle_post_action_dm(
    ctx: Context[Tux], user: Member | User, reason: str, action_desc: str
) -> bool

Handle DM sending after successful action execution.

Returns:

  • True if DM was sent, False otherwise –
_send_response_embed async
Python
_send_response_embed(
    ctx: Context[Tux], case: Case | None, user: Member | User, dm_sent: bool
) -> None

Send the response embed for the moderation action.

_send_mod_log_embed async
Python
_send_mod_log_embed(
    ctx: Context[Tux], case: Case, user: Member | User, dm_sent: bool
) -> Message | None

Send the response embed to the mod log channel.

_get_default_dm_action
Python
_get_default_dm_action(case_type: CaseType) -> str

Get the default DM action description for a case type.

Returns:

  • str –

    Default action description for the case type.