Skip to content

execution_service

Execution service for moderation operations.

Handles retry logic, circuit breakers, and execution management using proper service composition.

Classes:

  • ExecutionService –

    Service for executing moderation actions with retry logic.

Classes

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.