Skip to content

tux.database.controllers.levels

Classes:

Name Description
LevelsController

Controller for managing user levels and experience.

Classes

LevelsController()

Bases: BaseController[Levels]

Controller for managing user levels and experience.

This controller provides methods for tracking, updating, and querying user levels and experience points across guilds.

Initialize the LevelsController with the levels table.

Methods:

Name Description
get_xp

Get the XP of a member in a guild.
get_level

Get the level of a member in a guild.
get_xp_and_level

Get the XP and level of a member in a guild.
get_last_message_time

Get the last message time of a member in a guild.
is_blacklisted

Check if a member is blacklisted in a guild.
update_xp_and_level

Update the XP and level of a member in a guild.
toggle_blacklist

Toggle the blacklist status of a member in a guild.
find_one

Finds the first record matching specified criteria.
find_unique

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

Reset the XP and level of a member in a guild.
find_many

Finds multiple records matching specified criteria.
get_top_members

Get the top members in a guild by XP.
add_xp

Add XP to a member and calculate if they leveled up.
count

Counts records matching the specified criteria.
create

Creates a new record in the table.
calculate_level

Calculate level based on XP.
update

Updates a single existing record matching the criteria.
count_ranked_members

Count the number of ranked members in a guild.
get_rank

Get the rank of a member in a guild.
delete

Deletes a single record matching the criteria.
upsert

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

Delete all levels data for a guild.
update_many

Updates multiple records matching the criteria.
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/levels.py
Python
def __init__(self) -> None:
    """Initialize the LevelsController with the levels table."""
    super().__init__("levels")
    self.guild_table: GuildActions[Guild] = db.client.guild

Functions

get_xp(member_id: int, guild_id: int) -> float async

Get the XP of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
float

The XP of the member, or 0.0 if not found
Source code in tux/database/controllers/levels.py
Python
async def get_xp(self, member_id: int, guild_id: int) -> float:
    """Get the XP of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    float
        The XP of the member, or 0.0 if not found
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "xp", 0.0)
    except Exception as e:
        logger.error(f"Error querying XP for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return 0.0
get_level(member_id: int, guild_id: int) -> int async

Get the level of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

The level of the member, or 0 if not found
Source code in tux/database/controllers/levels.py
Python
async def get_level(self, member_id: int, guild_id: int) -> int:
    """Get the level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The level of the member, or 0 if not found
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "level", 0)
    except Exception as e:
        logger.error(f"Error querying level for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return 0
get_xp_and_level(member_id: int, guild_id: int) -> tuple[float, int] async

Get the XP and level of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
tuple[float, int]

A tuple containing the XP and level of the member, or (0.0, 0) if not found
Source code in tux/database/controllers/levels.py
Python
async def get_xp_and_level(self, member_id: int, guild_id: int) -> tuple[float, int]:
    """Get the XP and level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    tuple[float, int]
        A tuple containing the XP and level of the member, or (0.0, 0) if not found
    """
    try:
        record = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})

        if record:
            return (self.safe_get_attr(record, "xp", 0.0), self.safe_get_attr(record, "level", 0))
        return (0.0, 0)  # noqa: TRY300

    except Exception as e:
        logger.error(f"Error querying XP and level for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return (0.0, 0)
_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/levels.py
Python
    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    tuple[float, int]
        A tuple containing the XP and level of the member, or (0.0, 0) if not found
    """
    try:
        record = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})

        if record:
            return (self.safe_get_attr(record, "xp", 0.0), self.safe_get_attr(record, "level", 0))
        return (0.0, 0)  # noqa: TRY300

    except Exception as e:
        logger.error(f"Error querying XP and level for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return (0.0, 0)

async def get_last_message_time(self, member_id: int, guild_id: int) -> datetime.datetime | None:
    """Get the last message time of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    datetime.datetime | None
        The last message time of the member, or None if not found
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "last_message", None)
    except Exception as e:
        logger.error(f"Error querying last message time for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return None

async def is_blacklisted(self, member_id: int, guild_id: int) -> bool:
get_last_message_time(member_id: int, guild_id: int) -> datetime.datetime | None async

Get the last message time of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
datetime | None

The last message time of the member, or None if not found
Source code in tux/database/controllers/levels.py
Python
async def get_last_message_time(self, member_id: int, guild_id: int) -> datetime.datetime | None:
    """Get the last message time of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    datetime.datetime | None
        The last message time of the member, or None if not found
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "last_message", None)
    except Exception as e:
        logger.error(f"Error querying last message time for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return None
is_blacklisted(member_id: int, guild_id: int) -> bool async

Check if a member is blacklisted in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
bool

True if the member is blacklisted, False otherwise
Source code in tux/database/controllers/levels.py
Python
async def is_blacklisted(self, member_id: int, guild_id: int) -> bool:
    """Check if a member is blacklisted in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    bool
        True if the member is blacklisted, False otherwise
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "blacklisted", False)
    except Exception as e:
        logger.error(f"Error querying blacklist status for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return False
_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/levels.py
Python
Parameters
----------
member_id : int
_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/levels.py
Python
    guild_id : int
        The ID of the guild

    Returns
    -------
    bool
        True if the member is blacklisted, False otherwise
    """
    try:
        levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
        return self.safe_get_attr(levels, "blacklisted", False)
    except Exception as e:
        logger.error(f"Error querying blacklist status for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return False

async def update_xp_and_level(
    self,
    member_id: int,
    guild_id: int,
    xp: float,
    level: int,
update_xp_and_level(member_id: int, guild_id: int, xp: float, level: int, last_message: datetime.datetime) -> Levels | None async

Update the XP and level of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required
xp float

The XP of the member

 required
level int

The level of the member

 required
last_message datetime

The last message time of the member

 required

Returns:

Type Description
Levels | None

The updated levels record, or None if the update failed
Source code in tux/database/controllers/levels.py
Python
async def update_xp_and_level(
    self,
    member_id: int,
    guild_id: int,
    xp: float,
    level: int,
    last_message: datetime.datetime,
) -> Levels | None:
    """Update the XP and level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild
    xp : float
        The XP of the member
    level : int
        The level of the member
    last_message : datetime.datetime
        The last message time of the member

    Returns
    -------
    Levels | None
        The updated levels record, or None if the update failed
    """
    try:
        return await self.upsert(
            where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
            create={
                "member_id": member_id,
                "xp": xp,
                "level": level,
                "last_message": last_message,
                "guild": self.connect_or_create_relation("guild_id", guild_id),
            },
            update={"xp": xp, "level": level, "last_message": last_message},
        )
    except Exception as e:
        logger.error(f"Error updating XP and level for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return None
_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/levels.py
Python
) -> Levels | None:
    """Update the XP and level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild
    xp : float
_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/levels.py
Python
level : int
    The level of the member
last_message : datetime.datetime
    The last message time of the member

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/levels.py
Python
    The updated levels record, or None if the update failed
"""
try:
    return await self.upsert(
        where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
        create={
            "member_id": member_id,
            "xp": xp,
            "level": level,
            "last_message": last_message,
_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/levels.py
Python
        },
        update={"xp": xp, "level": level, "last_message": last_message},
    )
except Exception as e:
    logger.error(f"Error updating XP and level for member_id: {member_id}, guild_id: {guild_id}: {e}")
    return None
toggle_blacklist(member_id: int, guild_id: int) -> bool async

Toggle the blacklist status of a member in a guild.

This method uses a transaction to ensure atomicity.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
bool

The new blacklist status of the member
Source code in tux/database/controllers/levels.py
Python
async def toggle_blacklist(self, member_id: int, guild_id: int) -> bool:
    """Toggle the blacklist status of a member in a guild.

    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    bool
        The new blacklist status of the member
    """

    async def toggle_tx():
        try:
            levels = await self.find_one(where={"member_id": member_id, "guild_id": guild_id})

            if levels is None:
                # Create new record with blacklisted=True
                await self.create(
                    data={
                        "member_id": member_id,
                        "blacklisted": True,
                        "xp": 0.0,
                        "level": 0,
                        "guild": self.connect_or_create_relation("guild_id", guild_id),
                    },
                )
                return True

            # Toggle existing record's blacklisted status
            current_status = self.safe_get_attr(levels, "blacklisted", False)
            new_status = not current_status

            await self.update(
                where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
                data={"blacklisted": new_status},
            )

            return new_status  # noqa: TRY300
        except Exception as e:
            logger.error(f"Error toggling blacklist for member_id: {member_id}, guild_id: {guild_id}: {e}")
            return False

    return await self.execute_transaction(toggle_tx)
_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/levels.py
Python
"""Toggle the blacklist status of a member in a guild.

This method uses a transaction to ensure atomicity.

Parameters
----------
member_id : int
    The ID of the member
guild_id : int
    The ID of the guild

Returns
-------
bool
    The new blacklist status of the member
"""
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Levels | 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/levels.py
Python
    if levels is None:
        # Create new record with blacklisted=True
        await self.create(
            data={
                "member_id": member_id,
                "blacklisted": True,
                "xp": 0.0,
                "level": 0,
                "guild": self.connect_or_create_relation("guild_id", guild_id),
            },
        )
        return True

    # Toggle existing record's blacklisted status
    current_status = self.safe_get_attr(levels, "blacklisted", False)
    new_status = not current_status

    await self.update(
        where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
        data={"blacklisted": new_status},
    )

    return new_status  # noqa: TRY300
except Exception as e:
    logger.error(f"Error toggling blacklist for member_id: {member_id}, guild_id: {guild_id}: {e}")
    return False
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Levels | 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/levels.py
Python
    return await self.execute_transaction(toggle_tx)

async def reset_xp(self, member_id: int, guild_id: int) -> Levels | None:
    """Reset the XP and level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    Levels | None
        The updated levels record, or None if the update failed
    """
    try:
        result = await self.update(
            where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
            data={"xp": 0.0, "level": 0},
        )
    except Exception as e:
        logger.error(f"Error resetting XP for member_id: {member_id}, guild_id: {guild_id}: {e}")
reset_xp(member_id: int, guild_id: int) -> Levels | None async

Reset the XP and level of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
Levels | None

The updated levels record, or None if the update failed
Source code in tux/database/controllers/levels.py
Python
async def reset_xp(self, member_id: int, guild_id: int) -> Levels | None:
    """Reset the XP and level of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    Levels | None
        The updated levels record, or None if the update failed
    """
    try:
        result = await self.update(
            where={"member_id_guild_id": {"member_id": member_id, "guild_id": guild_id}},
            data={"xp": 0.0, "level": 0},
        )
    except Exception as e:
        logger.error(f"Error resetting XP for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return None
    else:
        return result
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[Levels] 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/levels.py
Python
    else:
        return result

async def get_top_members(self, guild_id: int, limit: int = 10, skip: int = 0) -> list[Levels]:
    """Get the top members in a guild by XP.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int
        The maximum number of members to return
    skip : int
        The number of members to skip

    Returns
    -------
    list[Levels]
        The top members in the guild by XP
    """
    try:
        return await self.find_many(
            where={"guild_id": guild_id, "blacklisted": False},
            order={"xp": "desc"},
            take=limit,
            skip=skip,
        )
    except Exception as e:
        logger.error(f"Error querying top members for guild_id: {guild_id}: {e}")
        return []

async def add_xp(self, member_id: int, guild_id: int, xp_to_add: float) -> tuple[float, int, bool]:
    """Add XP to a member and calculate if they leveled up.

    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild
    xp_to_add : float
get_top_members(guild_id: int, limit: int = 10, skip: int = 0) -> list[Levels] async

Get the top members in a guild by XP.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required
limit int

The maximum number of members to return

 10
skip int

The number of members to skip

 0

Returns:

Type Description
list[Levels]

The top members in the guild by XP
Source code in tux/database/controllers/levels.py
Python
async def get_top_members(self, guild_id: int, limit: int = 10, skip: int = 0) -> list[Levels]:
    """Get the top members in a guild by XP.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int
        The maximum number of members to return
    skip : int
        The number of members to skip

    Returns
    -------
    list[Levels]
        The top members in the guild by XP
    """
    try:
        return await self.find_many(
            where={"guild_id": guild_id, "blacklisted": False},
            order={"xp": "desc"},
            take=limit,
            skip=skip,
        )
    except Exception as e:
        logger.error(f"Error querying top members for guild_id: {guild_id}: {e}")
        return []
add_xp(member_id: int, guild_id: int, xp_to_add: float) -> tuple[float, int, bool] async

Add XP to a member and calculate if they leveled up.

This method uses a transaction to ensure atomicity.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required
xp_to_add float

The amount of XP to add

 required

Returns:

Type Description
tuple[float, int, bool]

A tuple containing the new XP, new level, and whether the member leveled up
Source code in tux/database/controllers/levels.py
Python
async def add_xp(self, member_id: int, guild_id: int, xp_to_add: float) -> tuple[float, int, bool]:
    """Add XP to a member and calculate if they leveled up.

    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild
    xp_to_add : float
        The amount of XP to add

    Returns
    -------
    tuple[float, int, bool]
        A tuple containing the new XP, new level, and whether the member leveled up
    """

    async def add_xp_tx():
        # Initialize with defaults in case of failure
        current_xp = 0.0
        current_level = 0

        try:
            # Get current XP and level
            current_xp, current_level = await self.get_xp_and_level(member_id, guild_id)

            # Calculate new XP and level
            new_xp = current_xp + xp_to_add
            new_level = self.calculate_level(new_xp)
            leveled_up = new_level > current_level

            # Update database
            now = datetime.datetime.now(datetime.UTC)
            await self.update_xp_and_level(
                member_id=member_id,
                guild_id=guild_id,
                xp=new_xp,
                level=new_level,
                last_message=now,
            )
        except Exception as e:
            logger.error(f"Error adding XP for member_id: {member_id}, guild_id: {guild_id}: {e}")
            return (current_xp, current_level, False)
        else:
            return (new_xp, new_level, leveled_up)

    return await self.execute_transaction(add_xp_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/levels.py
Python
Returns
-------
tuple[float, int, bool]
    A tuple containing the new XP, new level, and whether the member leveled up
"""

async def add_xp_tx():
    # Initialize with defaults in case of failure
    current_xp = 0.0
    current_level = 0

    try:
        # Get current XP and level
        current_xp, current_level = await self.get_xp_and_level(member_id, guild_id)

        # Calculate new XP and level
        new_xp = current_xp + xp_to_add
        new_level = self.calculate_level(new_xp)
        leveled_up = new_level > current_level
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Levels 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/levels.py
Python
            # Update database
            now = datetime.datetime.now(datetime.UTC)
            await self.update_xp_and_level(
                member_id=member_id,
                guild_id=guild_id,
                xp=new_xp,
                level=new_level,
                last_message=now,
            )
        except Exception as e:
            logger.error(f"Error adding XP for member_id: {member_id}, guild_id: {guild_id}: {e}")
            return (current_xp, current_level, False)
        else:
            return (new_xp, new_level, leveled_up)

    return await self.execute_transaction(add_xp_tx)

@staticmethod
def calculate_level(xp: float) -> int:
    """Calculate level based on XP.

    This uses a standard RPG-style level curve.

    Parameters
calculate_level(xp: float) -> int staticmethod

Calculate level based on XP.

This uses a standard RPG-style level curve.

Parameters:

Name Type Description Default
xp float

The XP to calculate the level from

 required

Returns:

Type Description
int

The calculated level
Source code in tux/database/controllers/levels.py
Python
@staticmethod
def calculate_level(xp: float) -> int:
    """Calculate level based on XP.

    This uses a standard RPG-style level curve.

    Parameters
    ----------
    xp : float
        The XP to calculate the level from

    Returns
    -------
    int
        The calculated level
    """
    # Base calculation: level = floor(sqrt(xp / 100))
    import math

    return math.floor(math.sqrt(xp / 100))
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Levels | 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/levels.py
Python
    xp : float
        The XP to calculate the level from

    Returns
    -------
    int
        The calculated level
    """
    # Base calculation: level = floor(sqrt(xp / 100))
    import math

    return math.floor(math.sqrt(xp / 100))

async def count_ranked_members(self, guild_id: int) -> int:
    """Count the number of ranked members in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of ranked members
    """
    return await self.count(where={"guild_id": guild_id, "blacklisted": False})
count_ranked_members(guild_id: int) -> int async

Count the number of ranked members in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

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

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of ranked members
    """
    return await self.count(where={"guild_id": guild_id, "blacklisted": False})
get_rank(member_id: int, guild_id: int) -> int async

Get the rank of a member in a guild.

Parameters:

Name Type Description Default
member_id int

The ID of the member

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

The rank of the member (1-based), or 0 if not found
Source code in tux/database/controllers/levels.py
Python
async def get_rank(self, member_id: int, guild_id: int) -> int:
    """Get the rank of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The rank of the member (1-based), or 0 if not found
    """
    try:
        # Get the member's XP
        member_xp = await self.get_xp(member_id, guild_id)

        # Count members with more XP
        higher_ranked = await self.count(
            where={
                "guild_id": guild_id,
                "blacklisted": False,
                "xp": {"gt": member_xp},
            },
        )

        # Rank is position (1-based)
        return higher_ranked + 1
    except Exception as e:
        logger.error(f"Error getting rank for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return 0
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Levels | 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/levels.py
Python
async def get_rank(self, member_id: int, guild_id: int) -> int:
    """Get the rank of a member in a guild.

    Parameters
    ----------
    member_id : int
        The ID of the member
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The rank of the member (1-based), or 0 if not found
    """
    try:
        # Get the member's XP
        member_xp = await self.get_xp(member_id, guild_id)

        # Count members with more XP
        higher_ranked = await self.count(
            where={
                "guild_id": guild_id,
                "blacklisted": False,
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Levels 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/levels.py
Python
            },
        )

        # Rank is position (1-based)
        return higher_ranked + 1
    except Exception as e:
        logger.error(f"Error getting rank for member_id: {member_id}, guild_id: {guild_id}: {e}")
        return 0

async def bulk_delete_by_guild_id(self, guild_id: int) -> int:
    """Delete all levels data for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

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

Delete all levels data for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

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

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of records deleted
    """
    return await self.delete_many(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.
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.