Skip to content

case

Moderation case management controller.

This controller manages moderation cases (bans, kicks, timeouts, etc.) with automatic case numbering, status tracking, and audit logging for Discord guilds.

Classes:

  • CaseController –

    Clean Case controller using the new BaseController pattern.

Classes

CaseController

Python
CaseController(db: DatabaseService | None = None)

Bases: BaseController[Case]

Clean Case controller using the new BaseController pattern.

Initialize the case controller.

Parameters:

  • db (DatabaseService | None, default: None ) –

    The database service instance. If None, uses the default service.

Returned by: Used by:

Methods:

Attributes:

Attributes

db_service property
Python
db_service: DatabaseService

Database service property for test compatibility.

Returns:

model_class property
Python
model_class: type[ModelT]

Model class property for test compatibility.

Returns:

  • type[ModelT] –

    The SQLModel class.

Functions

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

Get a case by its ID.

Returns:

  • Case | None –

    The case if found, None otherwise.

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

Get all cases for a specific user in a guild.

Returns:

  • list[Case] –

    List of all cases for the user in the guild.

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

Get all active cases for a specific user in a guild.

Returns:

  • list[Case] –

    List of active cases for the user in the guild.

create_case async
Python
create_case(
    case_type: str,
    case_user_id: int,
    case_moderator_id: int,
    guild_id: int,
    case_reason: str | None = None,
    case_status: bool = True,
    **kwargs: Any,
) -> Case

Create a new case with auto-generated case number.

Uses SELECT FOR UPDATE to prevent race conditions when generating case numbers.

Parameters:

  • case_type (str) –

    The type of case (from CaseType enum value)

  • case_user_id (int) –

    Discord ID of the user being moderated

  • case_moderator_id (int) –

    Discord ID of the moderator

  • guild_id (int) –

    Discord guild ID

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

    Reason for the moderation action

  • case_status (bool, default: True ) –

    Whether the case is active (default True)

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

    Additional case fields (e.g., case_expires_at, case_metadata, mod_log_message_id)

Returns:

  • Case –

    The newly created case with auto-generated case number.

Notes
  • For expiring cases, use case_expires_at (datetime) in kwargs
  • Do NOT pass duration - convert to case_expires_at before calling this method
  • Case numbers are auto-generated per guild using SELECT FOR UPDATE locking
_get_pagination
Python
_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

  • PaginationController[ModelT] –

    The pagination controller instance.

_get_bulk
Python
_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

  • BulkOperationsController[ModelT] –

    The bulk operations controller instance.

_get_transaction
Python
_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

  • TransactionController[ModelT] –

    The transaction controller instance.

_get_performance
Python
_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

  • PerformanceController[ModelT] –

    The performance controller instance.

_get_upsert
Python
_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

  • UpsertController[ModelT] –

    The upsert controller instance.

create async
Python
create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

  • ModelT –

    The newly created record.

get_by_id async
Python
get_by_id(record_id: Any) -> ModelT | None

Get a record by ID.

Returns:

  • ModelT | None –

    The record if found, None otherwise.

update_by_id async
Python
update_by_id(record_id: Any, **values: Any) -> ModelT | None

Update a record by ID.

Returns:

  • ModelT | None –

    The updated record, or None if not found.

update_case async
Python
update_case(case_id: int, **kwargs: Any) -> Case | None

Update a case by ID.

Returns:

  • Case | None –

    The updated case, or None if not found.

delete_by_id async
Python
delete_by_id(record_id: Any) -> bool

Delete a record by ID.

Returns:

  • bool –

    True if deleted successfully, False otherwise.

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.

exists async
Python
exists(filters: Any) -> bool

Check if a record exists.

Returns:

  • bool –

    True if record exists, False otherwise.

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

Close a case by setting its status to False.

Returns:

  • Case | None –

    The updated case, or None if not found.

find_one async
Python
find_one(filters: Any | None = None, order_by: Any | None = None) -> ModelT | None

Find one record.

Returns:

  • ModelT | None –

    The found record, or None if not found.

delete_case async
Python
delete_case(case_id: int) -> bool

Delete a case by ID.

Returns:

  • bool –

    True if deleted successfully, False otherwise.

find_all async
Python
find_all(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
) -> list[ModelT]

Find all records with performance optimizations.

Returns:

  • list[ModelT] –

    List of found records.

get_cases_by_guild async
Python
get_cases_by_guild(guild_id: int, limit: int | None = None) -> list[Case]

Get all cases for a guild, optionally limited.

Returns:

  • list[Case] –

    List of cases for the guild.

get_cases_by_type async
Python
get_cases_by_type(guild_id: int, case_type: str) -> list[Case]

Get all cases of a specific type in a guild.

Returns:

  • list[Case] –

    List of cases matching the specified type.

find_all_with_options async
Python
find_all_with_options(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
    load_relationships: list[str] | None = None,
) -> list[ModelT]

Find all records with relationship loading options.

Returns:

  • list[ModelT] –

    List of found records with loaded relationships.

get_recent_cases async
Python
get_recent_cases(guild_id: int, hours: int = 24) -> list[Case]

Get cases created within the last N hours.

Returns:

count async
Python
count(filters: Any | None = None) -> int

Count records.

Returns:

  • int –

    The count of matching records.

get_case_count_by_guild async
Python
get_case_count_by_guild(guild_id: int) -> int

Get the total number of cases in a guild.

Returns:

  • int –

    The total count of cases in the guild.

get_all async
Python
get_all(filters: Any | None = None, order_by: Any | None = None) -> list[ModelT]

Get all records (alias for find_all without pagination).

Returns:

  • list[ModelT] –

    List of all matching records.

is_user_under_restriction async
Python
is_user_under_restriction(
    user_id: int | None = None, guild_id: int | None = None, **kwargs: Any
) -> bool

Check if a user is under any active restriction in a guild.

Returns:

  • bool –

    True if user is under restriction, False otherwise.

execute_query async
Python
execute_query(query: Any) -> Any

Execute a custom query.

Returns:

  • Any –

    The query result.

find_with_json_query async
Python
find_with_json_query(
    json_column: str, json_path: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records using JSON column queries.

Returns:

  • list[ModelT] –

    List of records matching the JSON query.

get_case_by_number async
Python
get_case_by_number(case_number: int, guild_id: int) -> Case | None

Get a case by its case number in a guild.

Returns:

  • Case | None –

    The case if found, None otherwise.

find_with_array_contains async
Python
find_with_array_contains(
    array_column: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records where array column contains value.

Returns:

  • list[ModelT] –

    List of records with matching array values.

get_cases_by_options async
Python
get_cases_by_options(
    guild_id: int, options: dict[str, Any] | None = None
) -> list[Case]

Get cases by various filter options.

Returns:

  • list[Case] –

    List of cases matching the specified options.

find_with_full_text_search async
Python
find_with_full_text_search(
    search_columns: list[str], search_term: str, filters: Any | None = None
) -> list[ModelT]

Find records using full-text search.

Returns:

  • list[ModelT] –

    List of records matching the search term.

update_case_by_number async
Python
update_case_by_number(guild_id: int, case_number: int, **kwargs: Any) -> Case | None

Update a case by guild ID and case number.

Returns:

  • Case | None –

    The updated case, or None if not found.

paginate async
Python
paginate(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
) -> PaginationResult[ModelT]

Paginate records with metadata.

Returns:

  • PaginationResult[ModelT] –

    Pagination result with items, total, and page info.

get_all_cases async
Python
get_all_cases(guild_id: int) -> list[Case]

Get all cases in a guild.

Returns:

  • list[Case] –

    List of all cases in the guild.

find_paginated async
Python
find_paginated(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
    load_relationships: list[str] | None = None,
) -> PaginationResult[ModelT]

Find paginated records with relationship loading.

Returns:

  • PaginationResult[ModelT] –

    Pagination result with items and relationships loaded.

get_latest_case_by_user async
Python
get_latest_case_by_user(user_id: int, guild_id: int) -> Case | None

Get the most recent case for a user in a guild.

Returns:

  • Case | None –

    The most recent case if found, None otherwise.

bulk_create async
Python
bulk_create(items: list[dict[str, Any]]) -> list[ModelT]

Create multiple records in bulk.

Returns:

  • list[ModelT] –

    List of created records.

set_tempban_expired async
Python
set_tempban_expired(case_id: int, guild_id: int | None = None) -> bool

Mark a tempban case as processed after the user has been unbanned.

This sets case_processed=True to indicate the expiration has been handled. The case_status remains True (the case is still valid, just completed). The case_expires_at field remains unchanged as a historical record.

Parameters:

  • case_id (int) –

    The ID of the case to mark as processed

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

    Deprecated parameter kept for backward compatibility (unused)

Returns:

  • bool –

    True if the case was updated, False if not found

bulk_update async
Python
bulk_update(updates: list[tuple[Any, dict[str, Any]]]) -> int

Update multiple records in bulk.

Returns:

  • int –

    Number of records updated.

bulk_delete async
Python
bulk_delete(record_ids: list[Any]) -> int

Delete multiple records in bulk.

Returns:

  • int –

    Number of records deleted.

get_expired_tempbans async
Python
get_expired_tempbans(guild_id: int) -> list[Case]

Get tempban cases that have expired but haven't been processed yet.

Returns:

  • list[Case] –

    List of expired unprocessed tempban cases where case_expires_at is in the past, case_processed=False, and case_status=True.

update_where async
Python
update_where(filters: Any, values: dict[str, Any]) -> int

Update records matching filters.

Returns:

  • int –

    Number of records updated.

delete_where async
Python
delete_where(filters: Any) -> int

Delete records matching filters.

Returns:

  • int –

    Number of records deleted.

bulk_upsert_with_conflict_resolution async
Python
bulk_upsert_with_conflict_resolution(
    items: list[dict[str, Any]],
    conflict_columns: list[str],
    update_columns: list[str] | None = None,
) -> list[ModelT]

Bulk upsert with conflict resolution.

Returns:

  • list[ModelT] –

    List of upserted records.

get_case_count_by_user async
Python
get_case_count_by_user(user_id: int, guild_id: int) -> int

Get the total number of cases for a specific user in a guild.

Returns:

  • int –

    The total count of cases for the user.

with_session async
Python
with_session[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a session context.

Returns:

  • R –

    The result of the operation.

get_cases_by_moderator async
Python
get_cases_by_moderator(moderator_id: int, guild_id: int) -> list[Case]

Get all cases moderated by a specific user in a guild.

Returns:

  • list[Case] –

    List of cases moderated by the user.

with_transaction async
Python
with_transaction[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a transaction context.

Returns:

  • R –

    The result of the operation.

get_expired_cases async
Python
get_expired_cases(guild_id: int) -> list[Case]

Get all expired cases (any type) that haven't been processed yet.

Returns:

  • list[Case] –

    List of expired unprocessed cases where case_expires_at is in the past, case_processed=False, and case_status=True.

execute_transaction async
Python
execute_transaction(callback: Callable[[], Any]) -> Any

Execute a callback within a transaction.

Returns:

  • Any –

    The result of the callback.

get_table_statistics async
Python
get_table_statistics() -> dict[str, Any]

Get comprehensive table statistics.

Returns:

  • dict[str, Any] –

    Dictionary containing table statistics.

explain_query_performance async
Python
explain_query_performance(
    query: Any, analyze: bool = False, buffers: bool = False
) -> dict[str, Any]

Explain query performance with optional analysis.

Returns:

  • dict[str, Any] –

    Dictionary containing query execution plan and statistics.

upsert_by_field async
Python
upsert_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Upsert a record by a specific field.

Returns:

  • tuple[ModelT, bool] –

    Tuple of (record, created) where created is True if new record was created.

upsert_by_id async
Python
upsert_by_id(
    record_id: Any, defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record by ID.

Returns:

  • tuple[ModelT, bool] –

    Tuple of (record, created) where created is True if new record was created.

get_or_create_by_field async
Python
get_or_create_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Get existing record or create new one by field.

Returns:

  • tuple[ModelT, bool] –

    Tuple of (record, created) where created is True if new record was created.

get_or_create async
Python
get_or_create(
    defaults: dict[str, Any] | None = None, **filters: Any
) -> tuple[ModelT, bool]

Get existing record or create new one.

Returns:

  • tuple[ModelT, bool] –

    Tuple of (record, created) where created is True if new record was created.

upsert async
Python
upsert(
    filters: dict[str, Any], defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record.

Returns:

  • tuple[ModelT, bool] –

    Tuple of (record, created) where created is True if new record was created.