tux.database.controllers.afk
¶
Classes:
Name | Description |
---|---|
AfkController | Controller for managing AFK status records. |
Classes¶
AfkController()
¶
Bases: BaseController[AFKModel]
Controller for managing AFK status records.
This controller provides methods for tracking, checking, and managing AFK (Away From Keyboard) status of guild members.
Initialize the AfkController with the afkmodel table.
Methods:
Name | Description |
---|---|
get_afk_member | Get the AFK record for a member in a guild. |
is_afk | Check if a member is AFK in a guild. |
is_perm_afk | Check if a member is permanently AFK in a guild. |
set_afk | Insert or update an AFK record for a member. |
remove_afk | Remove an AFK record for a member. |
count_afk_members | Count the number of AFK members in a guild. |
get_all_afk_members | Get all AFK members 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). |
find_many | Finds multiple records matching specified criteria. |
count | Counts records matching the specified criteria. |
create | Creates a new record in the table. |
update | Updates a single existing record matching the criteria. |
delete | Deletes a single record matching the criteria. |
upsert | Updates a record if it exists, otherwise creates it. |
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/afk.py
Functions¶
get_afk_member(member_id: int, *, guild_id: int) -> AFKModel | None
async
¶
Get the AFK record for a member in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
member_id | int | The ID of the member to check | required |
guild_id | int | The ID of the guild to check in | required |
Returns:
Type | Description |
---|---|
AFKModel | None | The AFK record if found, None otherwise |
Source code in tux/database/controllers/afk.py
async def get_afk_member(self, member_id: int, *, guild_id: int) -> AFKModel | None:
"""Get the AFK record for a member in a guild.
Parameters
----------
member_id : int
The ID of the member to check
guild_id : int
The ID of the guild to check in
Returns
-------
AFKModel | None
The AFK record if found, None otherwise
"""
return await self.find_one(where={"member_id": member_id, "guild_id": guild_id})
is_afk(member_id: int, *, guild_id: int) -> bool
async
¶
Check if a member is AFK in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
member_id | int | The ID of the member to check | required |
guild_id | int | The ID of the guild to check in | required |
Returns:
Type | Description |
---|---|
bool | True if the member is AFK, False otherwise |
Source code in tux/database/controllers/afk.py
async def is_afk(self, member_id: int, *, guild_id: int) -> bool:
"""Check if a member is AFK in a guild.
Parameters
----------
member_id : int
The ID of the member to check
guild_id : int
The ID of the guild to check in
Returns
-------
bool
True if the member is AFK, False otherwise
"""
entry = await self.get_afk_member(member_id, guild_id=guild_id)
return entry is not None
is_perm_afk(member_id: int, *, guild_id: int) -> bool
async
¶
Check if a member is permanently AFK in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
member_id | int | The ID of the member to check | required |
guild_id | int | The ID of the guild to check in | required |
Returns:
Type | Description |
---|---|
bool | True if the member is permanently AFK, False otherwise |
Source code in tux/database/controllers/afk.py
async def is_perm_afk(self, member_id: int, *, guild_id: int) -> bool:
"""Check if a member is permanently AFK in a guild.
Parameters
----------
member_id : int
The ID of the member to check
guild_id : int
The ID of the guild to check in
Returns
-------
bool
True if the member is permanently AFK, False otherwise
"""
is_user_perm_afk = await self.find_one(
where={"member_id": member_id, "guild_id": guild_id, "perm_afk": True},
)
return is_user_perm_afk is not None
_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/afk.py
"""
is_user_perm_afk = await self.find_one(
where={"member_id": member_id, "guild_id": guild_id, "perm_afk": True},
)
return is_user_perm_afk is not None
async def set_afk(
self,
member_id: int,
nickname: str,
reason: str,
guild_id: int,
perm_afk: bool = False,
until: datetime | None = None,
enforced: bool = False,
) -> AFKModel:
"""Insert or update an AFK record for a member.
Parameters
----------
member_id : int
The ID of the member to set as AFK
nickname : str
The nickname of the member
reason : str
The reason for being AFK
guild_id : int
The ID of the guild
perm_afk : bool
Whether the AFK status is permanent
Returns
-------
AFKModel
The created or updated AFK record
"""
create_data = {
"member_id": member_id,
"nickname": nickname,
"reason": reason,
"perm_afk": perm_afk,
"guild": self.connect_or_create_relation("guild_id", guild_id),
"until": until,
"enforced": enforced,
"since": datetime.now(UTC),
}
set_afk(member_id: int, nickname: str, reason: str, guild_id: int, perm_afk: bool = False, until: datetime | None = None, enforced: bool = False) -> AFKModel
async
¶
Insert or update an AFK record for a member.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
member_id | int | The ID of the member to set as AFK | required |
nickname | str | The nickname of the member | required |
reason | str | The reason for being AFK | required |
guild_id | int | The ID of the guild | required |
perm_afk | bool | Whether the AFK status is permanent | False |
Returns:
Type | Description |
---|---|
AFKModel | The created or updated AFK record |
Source code in tux/database/controllers/afk.py
async def set_afk(
self,
member_id: int,
nickname: str,
reason: str,
guild_id: int,
perm_afk: bool = False,
until: datetime | None = None,
enforced: bool = False,
) -> AFKModel:
"""Insert or update an AFK record for a member.
Parameters
----------
member_id : int
The ID of the member to set as AFK
nickname : str
The nickname of the member
reason : str
The reason for being AFK
guild_id : int
The ID of the guild
perm_afk : bool
Whether the AFK status is permanent
Returns
-------
AFKModel
The created or updated AFK record
"""
create_data = {
"member_id": member_id,
"nickname": nickname,
"reason": reason,
"perm_afk": perm_afk,
"guild": self.connect_or_create_relation("guild_id", guild_id),
"until": until,
"enforced": enforced,
"since": datetime.now(UTC),
}
update_data = {
"nickname": nickname,
"reason": reason,
"perm_afk": perm_afk,
"until": until,
"enforced": enforced,
"since": datetime.now(UTC),
}
return await self.upsert(
where={"member_id": member_id},
create=create_data,
update=update_data,
include={"guild": True},
)
_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None
¶
_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/afk.py
"since": datetime.now(UTC),
}
return await self.upsert(
where={"member_id": member_id},
create=create_data,
update=update_data,
include={"guild": True},
)
async def remove_afk(self, member_id: int) -> AFKModel | None:
"""Remove an AFK record for a member.
Parameters
----------
member_id : int
The ID of the member to remove AFK status from
Returns
-------
AFKModel | None
remove_afk(member_id: int) -> AFKModel | None
async
¶
Remove an AFK record for a member.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
member_id | int | The ID of the member to remove AFK status from | required |
Returns:
Type | Description |
---|---|
AFKModel | None | The deleted AFK record if found, None otherwise |
Source code in tux/database/controllers/afk.py
async def remove_afk(self, member_id: int) -> AFKModel | None:
"""Remove an AFK record for a member.
Parameters
----------
member_id : int
The ID of the member to remove AFK status from
Returns
-------
AFKModel | None
The deleted AFK record if found, None otherwise
"""
return await self.delete(where={"member_id": member_id})
_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/afk.py
count_afk_members(guild_id: int) -> int
async
¶
Count the number of AFK members in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id | int | The ID of the guild to count AFK members for | required |
Returns:
Type | Description |
---|---|
int | The number of AFK members in the guild |
Source code in tux/database/controllers/afk.py
async def count_afk_members(self, guild_id: int) -> int:
"""Count the number of AFK members in a guild.
Parameters
----------
guild_id : int
The ID of the guild to count AFK members for
Returns
-------
int
The number of AFK members in the guild
"""
return await self.count(where={"guild_id": guild_id})
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
get_all_afk_members(guild_id: int) -> list[AFKModel]
async
¶
Get all AFK members in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id | int | The ID of the guild to get AFK members for | required |
Returns:
Type | Description |
---|---|
list[AFKModel] | List of AFK members in the guild |
Source code in tux/database/controllers/afk.py
async def get_all_afk_members(self, guild_id: int) -> list[AFKModel]:
"""Get all AFK members in a guild.
Parameters
----------
guild_id : int
The ID of the guild to get AFK members for
Returns
-------
list[AFKModel]
List of AFK members in the guild
"""
return await self.find_many(where={"guild_id": guild_id})
_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
_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.
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> AFKModel | 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. |
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel | 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. |
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[AFKModel]
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. |
count(where: dict[str, Any]) -> int
async
¶
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel
async
¶
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel | 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. |
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel | 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. |
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel
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. |
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 | 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. |