Skip to content

tux.database.controllers.case

Classes:

Name Description
CaseController

Controller for managing moderation cases.

Classes

CaseController()

Bases: BaseController[Case]

Controller for managing moderation cases.

This controller provides methods for creating, retrieving, updating, and deleting moderation cases in the database.

Initialize the CaseController with the case table.

Methods:

Name Description
get_next_case_number

Get the next case number for a guild.
insert_case

Insert a case into the database.
get_case_by_id

Get a case by its primary key ID.
get_all_cases

Get all cases for a guild.
get_cases_by_options

Get cases for a guild by options.
get_case_by_number

Get a case by its number in a guild.
get_all_cases_by_user_id

Get all cases for a target in a guild.
find_one

Finds the first record matching specified criteria.
get_all_cases_by_moderator_id

Get all cases for a moderator in a guild.
find_unique

Finds a single record by a unique constraint (e.g., ID).
get_latest_case_by_user

Get the latest case for a user with specified case types.
find_many

Finds multiple records matching specified criteria.
update_case

Update a case.
count

Counts records matching the specified criteria.
create

Creates a new record in the table.
delete_case_by_number

Delete a case by its number in a guild.
update

Updates a single existing record matching the criteria.
get_expired_tempbans

Get all cases that have expired tempbans.
delete

Deletes a single record matching the criteria.
set_tempban_expired

Set a tempban case as expired.
upsert

Updates a record if it exists, otherwise creates it.
bulk_delete_cases_by_guild_id

Delete all cases for a guild.
count_cases_by_guild_id

Count the number of cases in a guild.
update_many

Updates multiple records matching the criteria.
count_cases_by_user_id

Count the number of cases for a user in a guild.
delete_many

Deletes multiple records matching the criteria.
execute_transaction

Executes a series of database operations within a transaction.
connect_or_create_relation

Builds a Prisma 'connect_or_create' relation structure.
safe_get_attr

Safely retrieves an attribute from an object, returning a default if absent.
Source code in tux/database/controllers/case.py
Python
def __init__(self):
    """Initialize the CaseController with the case table."""
    super().__init__("case")
    # Access guild table through client property
    self.guild_table: GuildActions[Guild] = db.client.guild

Functions

get_next_case_number(guild_id: int) -> int async

Get the next case number for a guild.

This method automatically handles guild creation if it doesn't exist and atomically increments the case counter.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get the next case number for.

 required

Returns:

Type Description
int

The next case number for the guild.
Source code in tux/database/controllers/case.py
Python
async def get_next_case_number(self, guild_id: int) -> int:
    """Get the next case number for a guild.

    This method automatically handles guild creation if it doesn't exist
    and atomically increments the case counter.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get the next case number for.

    Returns
    -------
    int
        The next case number for the guild.
    """
    # Use connect_or_create to ensure guild exists and increment case count
    guild = await self.guild_table.upsert(
        where={"guild_id": guild_id},
        data={
            "create": {"guild_id": guild_id, "case_count": 1},
            "update": {"case_count": {"increment": 1}},
        },
    )

    return self.safe_get_attr(guild, "case_count", 1)
insert_case(guild_id: int, case_user_id: int, case_moderator_id: int, case_type: CaseType, case_reason: str, case_user_roles: list[int] | None = None, case_expires_at: datetime | None = None, case_tempban_expired: bool = False) -> Case async

Insert a case into the database.

This method automatically handles guild creation if needed using connect_or_create for optimal performance and race condition prevention.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to insert the case into.

 required
case_user_id int

The ID of the target of the case.

 required
case_moderator_id int

The ID of the moderator of the case.

 required
case_type CaseType

The type of the case.

 required
case_reason str

The reason for the case.

 required
case_user_roles list[int] | None

The roles of the target of the case.

 None
case_expires_at datetime | None

The expiration date of the case.

 None
case_tempban_expired bool

Whether the tempban has expired (Use only for tempbans).

 False

Returns:

Type Description
Case

The case database object.
Source code in tux/database/controllers/case.py
Python
async def insert_case(
    self,
    guild_id: int,
    case_user_id: int,
    case_moderator_id: int,
    case_type: CaseType,
    case_reason: str,
    case_user_roles: list[int] | None = None,
    case_expires_at: datetime | None = None,
    case_tempban_expired: bool = False,
) -> Case:
    """Insert a case into the database.

    This method automatically handles guild creation if needed using
    connect_or_create for optimal performance and race condition prevention.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to insert the case into.
    case_user_id : int
        The ID of the target of the case.
    case_moderator_id : int
        The ID of the moderator of the case.
    case_type : CaseType
        The type of the case.
    case_reason : str
        The reason for the case.
    case_user_roles : list[int] | None
        The roles of the target of the case.
    case_expires_at : datetime | None
        The expiration date of the case.
    case_tempban_expired : bool
        Whether the tempban has expired (Use only for tempbans).

    Returns
    -------
    Case
        The case database object.
    """
    case_number = await self.get_next_case_number(guild_id)

    # Create case with relation to guild using connect_or_create
    return await self.create(
        data={
            "case_number": case_number,
            "case_user_id": case_user_id,
            "case_moderator_id": case_moderator_id,
            "case_type": case_type,
            "case_reason": case_reason,
            "case_expires_at": case_expires_at,
            "case_user_roles": case_user_roles if case_user_roles is not None else [],
            "case_tempban_expired": case_tempban_expired,
            "guild": self.connect_or_create_relation("guild_id", guild_id),
        },
        include={"guild": True},
    )
_execute_query(operation: Callable[[], Any], error_msg: str) -> Any async

Executes a database query with standardized error logging.

Wraps the Prisma client operation call in a try-except block, logging any exceptions with a contextual error message.

Parameters:

Name Type Description Default
operation Callable[[], Any]

A zero-argument function (e.g., a lambda) that performs the database call.

 required
error_msg str

The base error message to log if an exception occurs.

 required

Returns:

Type Description
Any

The result of the database operation.

Raises:

Type Description
Exception

Re-raises any exception caught during the database operation.
Source code in tux/database/controllers/case.py
Python
    guild_id : int
        The ID of the guild to insert the case into.
    case_user_id : int
        The ID of the target of the case.
    case_moderator_id : int
        The ID of the moderator of the case.
    case_type : CaseType
        The type of the case.
    case_reason : str
        The reason for the case.
    case_user_roles : list[int] | None
        The roles of the target of the case.
    case_expires_at : datetime | None
        The expiration date of the case.
    case_tempban_expired : bool
        Whether the tempban has expired (Use only for tempbans).

    Returns
    -------
    Case
        The case database object.
    """
    case_number = await self.get_next_case_number(guild_id)

    # Create case with relation to guild using connect_or_create
    return await self.create(
        data={
            "case_number": case_number,
            "case_user_id": case_user_id,
            "case_moderator_id": case_moderator_id,
            "case_type": case_type,
            "case_reason": case_reason,
            "case_expires_at": case_expires_at,
            "case_user_roles": case_user_roles if case_user_roles is not None else [],
            "case_tempban_expired": case_tempban_expired,
            "guild": self.connect_or_create_relation("guild_id", guild_id),
        },
        include={"guild": True},
    )

async def get_case_by_id(self, case_id: int, include_guild: bool = False) -> Case | None:
    """Get a case by its primary key ID.

    Parameters
    ----------
    case_id : int
get_case_by_id(case_id: int, include_guild: bool = False) -> Case | None async

Get a case by its primary key ID.

Parameters:

Name Type Description Default
case_id int

The primary key ID of the case

 required
include_guild bool

Whether to include the guild relation

 False

Returns:

Type Description
Case | None

The case if found, otherwise None
Source code in tux/database/controllers/case.py
Python
async def get_case_by_id(self, case_id: int, include_guild: bool = False) -> Case | None:
    """Get a case by its primary key ID.

    Parameters
    ----------
    case_id : int
        The primary key ID of the case
    include_guild : bool
        Whether to include the guild relation

    Returns
    -------
    Case | None
        The case if found, otherwise None
    """
    include = {"guild": True} if include_guild else None
    return await self.find_unique(where={"case_id": case_id}, include=include)
_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None

Adds the 'include' argument to a dictionary if it is not None.

Source code in tux/database/controllers/case.py
Python
include_guild : bool
    Whether to include the guild relation

Returns
_build_find_args(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> dict[str, Any]

Constructs the keyword arguments dictionary for Prisma find operations.

Source code in tux/database/controllers/case.py
Python
    Case | None
        The case if found, otherwise None
    """
    include = {"guild": True} if include_guild else None
    return await self.find_unique(where={"case_id": case_id}, include=include)

async def get_all_cases(self, guild_id: int) -> list[Case]:
    """Get all cases for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get cases for.

    Returns
    -------
    list[Case]
        A list of cases for the guild.
    """
    return await self.find_many(
        where={"guild_id": guild_id},
get_all_cases(guild_id: int) -> list[Case] async

Get all cases for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get cases for.

 required

Returns:

Type Description
list[Case]

A list of cases for the guild.
Source code in tux/database/controllers/case.py
Python
async def get_all_cases(self, guild_id: int) -> list[Case]:
    """Get all cases for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get cases for.

    Returns
    -------
    list[Case]
        A list of cases for the guild.
    """
    return await self.find_many(
        where={"guild_id": guild_id},
        order={"case_created_at": "desc"},
    )
_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs simple keyword arguments for Prisma (e.g., create, delete).

Source code in tux/database/controllers/case.py
Python
    )

async def get_cases_by_options(
    self,
    guild_id: int,
    options: CaseWhereInput,
) -> list[Case]:
    """Get cases for a guild by options.

    Parameters
get_cases_by_options(guild_id: int, options: CaseWhereInput) -> list[Case] async

Get cases for a guild by options.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get cases for.

 required
options CaseWhereInput

The options to filter cases by.

 required

Returns:

Type Description
list[Case]

A list of cases for the guild matching the criteria.
Source code in tux/database/controllers/case.py
Python
async def get_cases_by_options(
    self,
    guild_id: int,
    options: CaseWhereInput,
) -> list[Case]:
    """Get cases for a guild by options.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get cases for.
    options : CaseWhereInput
        The options to filter cases by.

    Returns
    -------
    list[Case]
        A list of cases for the guild matching the criteria.
    """
    return await self.find_many(where={"guild_id": guild_id, **options}, order={"case_created_at": "desc"})
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for Prisma create operations.

Source code in tux/database/controllers/case.py
Python
guild_id : int
    The ID of the guild to get cases for.
options : CaseWhereInput
    The options to filter cases by.

Returns
-------
_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for Prisma update operations.

Source code in tux/database/controllers/case.py
Python
        A list of cases for the guild matching the criteria.
    """
    return await self.find_many(where={"guild_id": guild_id, **options}, order={"case_created_at": "desc"})

async def get_case_by_number(self, guild_id: int, case_number: int, include_guild: bool = False) -> Case | None:
    """Get a case by its number in a guild.

    Parameters
    ----------
    guild_id : int
get_case_by_number(guild_id: int, case_number: int, include_guild: bool = False) -> Case | None async

Get a case by its number in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get the case in.

 required
case_number int

The number of the case to get.

 required
include_guild bool

Whether to include the guild relation

 False

Returns:

Type Description
Case | None

The case if found, otherwise None.
Source code in tux/database/controllers/case.py
Python
async def get_case_by_number(self, guild_id: int, case_number: int, include_guild: bool = False) -> Case | None:
    """Get a case by its number in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get the case in.
    case_number : int
        The number of the case to get.
    include_guild : bool
        Whether to include the guild relation

    Returns
    -------
    Case | None
        The case if found, otherwise None.
    """
    include = {"guild": True} if include_guild else None
    return await self.find_one(where={"guild_id": guild_id, "case_number": case_number}, include=include)
_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for Prisma delete operations.

Source code in tux/database/controllers/case.py
Python
case_number : int
    The number of the case to get.
include_guild : bool
    Whether to include the guild relation

Returns
-------
_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for Prisma upsert operations.

Source code in tux/database/controllers/case.py
Python
        The case if found, otherwise None.
    """
    include = {"guild": True} if include_guild else None
    return await self.find_one(where={"guild_id": guild_id, "case_number": case_number}, include=include)

async def get_all_cases_by_user_id(
    self,
    guild_id: int,
    case_user_id: int,
    limit: int | None = None,
    include_guild: bool = False,
) -> list[Case]:
    """Get all cases for a target in a guild.

    Parameters
    ----------
    guild_id : int
get_all_cases_by_user_id(guild_id: int, case_user_id: int, limit: int | None = None, include_guild: bool = False) -> list[Case] async

Get all cases for a target in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get cases for.

 required
case_user_id int

The ID of the target to get cases for.

 required
limit int | None

Optional limit on the number of cases to return

 None
include_guild bool

Whether to include the guild relation

 False

Returns:

Type Description
list[Case]

A list of cases for the target in the guild.
Source code in tux/database/controllers/case.py
Python
async def get_all_cases_by_user_id(
    self,
    guild_id: int,
    case_user_id: int,
    limit: int | None = None,
    include_guild: bool = False,
) -> list[Case]:
    """Get all cases for a target in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get cases for.
    case_user_id : int
        The ID of the target to get cases for.
    limit : int | None
        Optional limit on the number of cases to return
    include_guild : bool
        Whether to include the guild relation

    Returns
    -------
    list[Case]
        A list of cases for the target in the guild.
    """
    include = {"guild": True} if include_guild else None
    return await self.find_many(
        where={"guild_id": guild_id, "case_user_id": case_user_id},
        include=include,
        take=limit,
        order={"case_created_at": "desc"},
    )
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Case | None async

Finds the first record matching specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required
include dict[str, bool]

Specifies relations to include in the result.

 None
order dict[str, str]

Specifies the field and direction for ordering.

 None

Returns:

Type Description
ModelType | None

The found record or None if no match exists.
Source code in tux/database/controllers/case.py
Python
    limit : int | None
        Optional limit on the number of cases to return
    include_guild : bool
        Whether to include the guild relation

    Returns
    -------
    list[Case]
        A list of cases for the target in the guild.
    """
    include = {"guild": True} if include_guild else None
    return await self.find_many(
        where={"guild_id": guild_id, "case_user_id": case_user_id},
        include=include,
        take=limit,
        order={"case_created_at": "desc"},
    )

async def get_all_cases_by_moderator_id(
    self,
    guild_id: int,
    case_moderator_id: int,
    limit: int | None = None,
) -> list[Case]:
    """Get all cases for a moderator in a guild.

    Parameters
get_all_cases_by_moderator_id(guild_id: int, case_moderator_id: int, limit: int | None = None) -> list[Case] async

Get all cases for a moderator in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get cases for.

 required
case_moderator_id int

The ID of the moderator to get cases for.

 required
limit int | None

Optional limit on the number of cases to return

 None

Returns:

Type Description
list[Case]

A list of cases for the moderator in the guild.
Source code in tux/database/controllers/case.py
Python
async def get_all_cases_by_moderator_id(
    self,
    guild_id: int,
    case_moderator_id: int,
    limit: int | None = None,
) -> list[Case]:
    """Get all cases for a moderator in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get cases for.
    case_moderator_id : int
        The ID of the moderator to get cases for.
    limit : int | None
        Optional limit on the number of cases to return

    Returns
    -------
    list[Case]
        A list of cases for the moderator in the guild.
    """
    return await self.find_many(
        where={"guild_id": guild_id, "case_moderator_id": case_moderator_id},
        take=limit,
        order={"case_created_at": "desc"},
    )
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Case | None async

Finds a single record by a unique constraint (e.g., ID).

Parameters:

Name Type Description Default
where dict[str, Any]

Unique query conditions (e.g., {'id': 1}).

 required
include dict[str, bool]

Specifies relations to include in the result.

 None

Returns:

Type Description
ModelType | None

The found record or None if no match exists.
Source code in tux/database/controllers/case.py
Python
    guild_id : int
        The ID of the guild to get cases for.
    case_moderator_id : int
        The ID of the moderator to get cases for.
    limit : int | None
        Optional limit on the number of cases to return

    Returns
    -------
    list[Case]
        A list of cases for the moderator in the guild.
    """
    return await self.find_many(
        where={"guild_id": guild_id, "case_moderator_id": case_moderator_id},
        take=limit,
        order={"case_created_at": "desc"},
    )

async def get_latest_case_by_user(
    self,
    guild_id: int,
    user_id: int,
    case_types: list[CaseType],
) -> Case | None:
get_latest_case_by_user(guild_id: int, user_id: int, case_types: list[CaseType]) -> Case | None async

Get the latest case for a user with specified case types.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get the case in.

 required
user_id int

The ID of the user to get the case for.

 required
case_types list[CaseType]

The types of cases to search for.

 required

Returns:

Type Description
Case | None

The latest case if found, otherwise None.
Source code in tux/database/controllers/case.py
Python
async def get_latest_case_by_user(
    self,
    guild_id: int,
    user_id: int,
    case_types: list[CaseType],
) -> Case | None:
    """Get the latest case for a user with specified case types.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get the case in.
    user_id : int
        The ID of the user to get the case for.
    case_types : list[CaseType]
        The types of cases to search for.

    Returns
    -------
    Case | None
        The latest case if found, otherwise None.
    """

    # Using a transaction to ensure read consistency
    async def get_latest_case():
        cases = await self.find_many(
            where={"guild_id": guild_id, "case_user_id": user_id},
            order={"case_created_at": "desc"},
            take=1,
        )

        if not cases:
            return None

        case = cases[0]
        case_type = self.safe_get_attr(case, "case_type")

        return case if case_type in case_types else None

    return await self.execute_transaction(get_latest_case)
find_many(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> list[Case] async

Finds multiple records matching specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required
include dict[str, bool]

Specifies relations to include in the results.

 None
order dict[str, str]

Specifies the field and direction for ordering.

 None
take int

Maximum number of records to return.

 None
skip int

Number of records to skip (for pagination).

 None
cursor dict[str, Any]

Cursor for pagination based on a unique field.

 None

Returns:

Type Description
list[ModelType]

A list of found records, potentially empty.
Source code in tux/database/controllers/case.py
Python
    Parameters
    ----------
    guild_id : int
        The ID of the guild to get the case in.
    user_id : int
        The ID of the user to get the case for.
    case_types : list[CaseType]
        The types of cases to search for.

    Returns
    -------
    Case | None
        The latest case if found, otherwise None.
    """

    # Using a transaction to ensure read consistency
    async def get_latest_case():
        cases = await self.find_many(
            where={"guild_id": guild_id, "case_user_id": user_id},
            order={"case_created_at": "desc"},
            take=1,
        )

        if not cases:
            return None

        case = cases[0]
        case_type = self.safe_get_attr(case, "case_type")

        return case if case_type in case_types else None

    return await self.execute_transaction(get_latest_case)

async def update_case(
    self,
    guild_id: int,
    case_number: int,
    case_reason: str,
    case_status: bool | None = None,
) -> Case | None:
    """Update a case.
update_case(guild_id: int, case_number: int, case_reason: str, case_status: bool | None = None) -> Case | None async

Update a case.

This method uses a transaction to ensure atomicity of the lookup and update.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to update the case in.

 required
case_number int

The number of the case to update.

 required
case_reason str

The new reason for the case.

 required
case_status bool | None

The new status for the case.

 None

Returns:

Type Description
Case | None

The updated case if found, otherwise None.
Source code in tux/database/controllers/case.py
Python
async def update_case(
    self,
    guild_id: int,
    case_number: int,
    case_reason: str,
    case_status: bool | None = None,
) -> Case | None:
    """Update a case.

    This method uses a transaction to ensure atomicity of the lookup and update.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to update the case in.
    case_number : int
        The number of the case to update.
    case_reason : str
        The new reason for the case.
    case_status : bool | None
        The new status for the case.

    Returns
    -------
    Case | None
        The updated case if found, otherwise None.
    """

    # Use a transaction to ensure the lookup and update are atomic
    async def update_case_tx():
        case = await self.find_one(where={"guild_id": guild_id, "case_number": case_number})
        if case is None:
            return None

        case_id = self.safe_get_attr(case, "case_id")
        update_data: dict[str, Any] = {"case_reason": case_reason}

        if case_status is not None:
            update_data["case_status"] = case_status

        return await self.update(where={"case_id": case_id}, data=update_data)

    return await self.execute_transaction(update_case_tx)
count(where: dict[str, Any]) -> int async

Counts records matching the specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required

Returns:

Type Description
int

The total number of matching records.
Source code in tux/database/controllers/case.py
Python
Parameters
----------
guild_id : int
    The ID of the guild to update the case in.
case_number : int
    The number of the case to update.
case_reason : str
    The new reason for the case.
case_status : bool | None
    The new status for the case.

Returns
-------
Case | None
    The updated case if found, otherwise None.
"""

# Use a transaction to ensure the lookup and update are atomic
async def update_case_tx():
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Case async

Creates a new record in the table.

Parameters:

Name Type Description Default
data dict[str, Any]

The data for the new record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The newly created record.
Source code in tux/database/controllers/case.py
Python
        if case is None:
            return None

        case_id = self.safe_get_attr(case, "case_id")
        update_data: dict[str, Any] = {"case_reason": case_reason}

        if case_status is not None:
            update_data["case_status"] = case_status

        return await self.update(where={"case_id": case_id}, data=update_data)

    return await self.execute_transaction(update_case_tx)

async def delete_case_by_number(self, guild_id: int, case_number: int) -> Case | None:
    """Delete a case by its number in a guild.

    This method uses a transaction to ensure atomicity of the lookup and delete.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete the case in.
    case_number : int
        The number of the case to delete.
delete_case_by_number(guild_id: int, case_number: int) -> Case | None async

Delete a case by its number in a guild.

This method uses a transaction to ensure atomicity of the lookup and delete.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to delete the case in.

 required
case_number int

The number of the case to delete.

 required

Returns:

Type Description
Case | None

The case if found and deleted, otherwise None.
Source code in tux/database/controllers/case.py
Python
async def delete_case_by_number(self, guild_id: int, case_number: int) -> Case | None:
    """Delete a case by its number in a guild.

    This method uses a transaction to ensure atomicity of the lookup and delete.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete the case in.
    case_number : int
        The number of the case to delete.

    Returns
    -------
    Case | None
        The case if found and deleted, otherwise None.
    """

    # Use a transaction to ensure the lookup and delete are atomic
    async def delete_case_tx():
        case = await self.find_one(where={"guild_id": guild_id, "case_number": case_number})
        if case is None:
            return None

        case_id = self.safe_get_attr(case, "case_id")
        return await self.delete(where={"case_id": case_id})

    return await self.execute_transaction(delete_case_tx)
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Case | None async

Updates a single existing record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to update.

 required
data dict[str, Any]

The data to update the record with.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType | None

The updated record, or None if no matching record was found.
Source code in tux/database/controllers/case.py
Python
    Returns
    -------
    Case | None
        The case if found and deleted, otherwise None.
    """

    # Use a transaction to ensure the lookup and delete are atomic
    async def delete_case_tx():
        case = await self.find_one(where={"guild_id": guild_id, "case_number": case_number})
        if case is None:
            return None

        case_id = self.safe_get_attr(case, "case_id")
        return await self.delete(where={"case_id": case_id})

    return await self.execute_transaction(delete_case_tx)

async def get_expired_tempbans(self) -> list[Case]:
    """Get all cases that have expired tempbans.

    Returns
    -------
    list[Case]
        A list of cases with expired tempbans.
    """
    return await self.find_many(
        where={
get_expired_tempbans() -> list[Case] async

Get all cases that have expired tempbans.

Returns:

Type Description
list[Case]

A list of cases with expired tempbans.
Source code in tux/database/controllers/case.py
Python
async def get_expired_tempbans(self) -> list[Case]:
    """Get all cases that have expired tempbans.

    Returns
    -------
    list[Case]
        A list of cases with expired tempbans.
    """
    return await self.find_many(
        where={
            "case_type": CaseType.TEMPBAN,
            "case_expires_at": {"lt": datetime.now(UTC)},
            "case_tempban_expired": False,
        },
    )
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Case | None async

Deletes a single record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to delete.

 required
include dict[str, bool]

Specifies relations to include in the returned deleted record.

 None

Returns:

Type Description
ModelType | None

The deleted record, or None if no matching record was found.
Source code in tux/database/controllers/case.py
Python
            "case_expires_at": {"lt": datetime.now(UTC)},
            "case_tempban_expired": False,
        },
    )

async def set_tempban_expired(self, case_number: int | None, guild_id: int) -> int | None:
    """Set a tempban case as expired.

    Parameters
    ----------
    case_number : int | None
        The number of the case to update.
    guild_id : int
        The ID of the guild the case belongs to.

    Returns
    -------
    int | None
        The number of Case records updated (1) if successful, None if no records were found,
        or raises an exception if multiple records were affected.
    """
    if case_number is None:
        msg = "Case number not found"
        raise ValueError(msg)
set_tempban_expired(case_number: int | None, guild_id: int) -> int | None async

Set a tempban case as expired.

Parameters:

Name Type Description Default
case_number int | None

The number of the case to update.

 required
guild_id int

The ID of the guild the case belongs to.

 required

Returns:

Type Description
int | None

The number of Case records updated (1) if successful, None if no records were found, or raises an exception if multiple records were affected.
Source code in tux/database/controllers/case.py
Python
async def set_tempban_expired(self, case_number: int | None, guild_id: int) -> int | None:
    """Set a tempban case as expired.

    Parameters
    ----------
    case_number : int | None
        The number of the case to update.
    guild_id : int
        The ID of the guild the case belongs to.

    Returns
    -------
    int | None
        The number of Case records updated (1) if successful, None if no records were found,
        or raises an exception if multiple records were affected.
    """
    if case_number is None:
        msg = "Case number not found"
        raise ValueError(msg)

    result = await self.update_many(
        where={"case_number": case_number, "guild_id": guild_id},
        data={"case_tempban_expired": True},
    )

    if result == 1:
        return result
    if result == 0:
        return None

    msg = f"Multiple records ({result}) were affected when updating case {case_number} in guild {guild_id}"
    raise ValueError(msg)
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Case async

Updates a record if it exists, otherwise creates it.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the existing record.

 required
create dict[str, Any]

Data to use if creating a new record.

 required
update dict[str, Any]

Data to use if updating an existing record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The created or updated record.
Source code in tux/database/controllers/case.py
Python
    result = await self.update_many(
        where={"case_number": case_number, "guild_id": guild_id},
        data={"case_tempban_expired": True},
    )

    if result == 1:
        return result
    if result == 0:
        return None

    msg = f"Multiple records ({result}) were affected when updating case {case_number} in guild {guild_id}"
    raise ValueError(msg)

async def bulk_delete_cases_by_guild_id(self, guild_id: int) -> int:
    """Delete all cases for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete cases for

    Returns
    -------
    int
        The number of cases deleted
    """
    return await self.delete_many(where={"guild_id": guild_id})

async def count_cases_by_guild_id(self, guild_id: int) -> int:
    """Count the number of cases in a guild.
bulk_delete_cases_by_guild_id(guild_id: int) -> int async

Delete all cases for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to delete cases for

 required

Returns:

Type Description
int

The number of cases deleted
Source code in tux/database/controllers/case.py
Python
async def bulk_delete_cases_by_guild_id(self, guild_id: int) -> int:
    """Delete all cases for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete cases for

    Returns
    -------
    int
        The number of cases deleted
    """
    return await self.delete_many(where={"guild_id": guild_id})
count_cases_by_guild_id(guild_id: int) -> int async

Count the number of cases in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to count cases for

 required

Returns:

Type Description
int

The number of cases in the guild
Source code in tux/database/controllers/case.py
Python
async def count_cases_by_guild_id(self, guild_id: int) -> int:
    """Count the number of cases in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to count cases for

    Returns
    -------
    int
        The number of cases in the guild
    """
    return await self.count(where={"guild_id": guild_id})
update_many(where: dict[str, Any], data: dict[str, Any]) -> int async

Updates multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to update.

 required
data dict[str, Any]

The data to update the records with.

 required

Returns:

Type Description
int

The number of records updated.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
Source code in tux/database/controllers/case.py
Python
    Parameters
    ----------
    guild_id : int
        The ID of the guild to count cases for

    Returns
    -------
    int
        The number of cases in the guild
    """
    return await self.count(where={"guild_id": guild_id})

async def count_cases_by_user_id(self, guild_id: int, user_id: int) -> int:
    """Count the number of cases for a user in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to count cases for
    user_id : int
        The ID of the user to count cases for

    Returns
    -------
    int
        The number of cases for the user in the guild
    """
    return await self.count(where={"guild_id": guild_id, "case_user_id": user_id})
count_cases_by_user_id(guild_id: int, user_id: int) -> int async

Count the number of cases for a user in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to count cases for

 required
user_id int

The ID of the user to count cases for

 required

Returns:

Type Description
int

The number of cases for the user in the guild
Source code in tux/database/controllers/case.py
Python
async def count_cases_by_user_id(self, guild_id: int, user_id: int) -> int:
    """Count the number of cases for a user in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to count cases for
    user_id : int
        The ID of the user to count cases for

    Returns
    -------
    int
        The number of cases for the user in the guild
    """
    return await self.count(where={"guild_id": guild_id, "case_user_id": user_id})
delete_many(where: dict[str, Any]) -> int async

Deletes multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to delete.

 required

Returns:

Type Description
int

The number of records deleted.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
execute_transaction(callback: Callable[[], Any]) -> Any async

Executes a series of database operations within a transaction.

Ensures atomicity: all operations succeed or all fail and roll back. Note: Does not use _execute_query internally to preserve specific transaction context in error messages.

Parameters:

Name Type Description Default
callback Callable[[], Any]

An async function containing the database operations to execute.

 required

Returns:

Type Description
Any

The result returned by the callback function.

Raises:

Type Description
Exception

Re-raises any exception that occurs during the transaction.
connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod

Builds a Prisma 'connect_or_create' relation structure.

Simplifies linking or creating related records during create/update operations.

Parameters:

Name Type Description Default
id_field str

The name of the ID field used for connection (e.g., 'guild_id').

 required
model_id Any

The ID value of the record to connect to.

 required
create_data dict[str, Any]

Additional data required if creating the related record. Must include at least the id_field and model_id.

 None

Returns:

Type Description
dict[str, Any]

A dictionary formatted for Prisma's connect_or_create.
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Safely retrieves an attribute from an object, returning a default if absent.

Parameters:

Name Type Description Default
obj Any

The object to retrieve the attribute from.

 required
attr str

The name of the attribute.

 required
default Any

The value to return if the attribute is not found. Defaults to None.

 None

Returns:

Type Description
Any

The attribute's value or the default value.