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 | Find the first matching record using SQLModel select(). |
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 SQLModel '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
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 SQLModel find operations.
Source code in tux/database/controllers/afk.py
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,
_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
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_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]
¶
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_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 SQLModel upsert operations.
Source code in tux/database/controllers/afk.py
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})
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
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})
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> ModelType | None
async
¶
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | 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[ModelType]
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) -> ModelType
async
¶
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | 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) -> ModelType | 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) -> ModelType
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 SQLModel '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 SQLModel'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. |