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. |
find_one | Return the first row that matches where or None. |
get_last_message_time | Get the last message time of a member in a guild. |
find_many | Return a list of rows matching where (or all rows). |
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. |
execute_transaction | Execute callback inside a database session / transaction block. |
reset_xp | Reset the XP and level of a member in a guild. |
safe_get_attr | Return |
connect_or_create_relation | Return a dict with a single key that can be merged into data dicts. |
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. |
calculate_level | Calculate level based on XP. |
count_ranked_members | Count the number of ranked members in a guild. |
get_rank | Get the rank of a member in a guild. |
bulk_delete_by_guild_id | Delete all levels data for a guild. |
Source code in tux/database/controllers/levels.py
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
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
_execute_query(op: Callable[[AsyncSession], Any], span_desc: str) -> Any
async
¶
Run op inside a managed session & sentry span (if enabled).
Source code in tux/database/controllers/levels.py
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)
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
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
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)
find_one(*, where: dict[str, Any], include: dict[str, bool] | None = None, **__: Any) -> ModelT | None
async
¶
Return the first row that matches where or None.
Source code in tux/database/controllers/levels.py
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
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
find_many(*, where: dict[str, Any] | None = None, include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None) -> list[ModelT]
async
¶
Return a list of rows matching where (or all rows).
Source code in tux/database/controllers/levels.py
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:
"""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
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
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
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
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
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
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)
execute_transaction(callback: Callable[[], Any]) -> Any
async
¶
Execute callback inside a database session / transaction block.
Source code in tux/database/controllers/levels.py
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)
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
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
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any
staticmethod
¶
connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any]
staticmethod
¶
Return a dict with a single key that can be merged into data dicts.
The calling code does something like::
data = {"guild": connect_or_create_relation("guild_id", guild_id)}
We map that pattern to a very small helper that collapses to {"guild_id": guild_id}
.
Source code in tux/database/controllers/levels.py
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
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
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
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)
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
@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))
return math.floor(math.sqrt(xp / 100))
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
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
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