Skip to content

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
Python
def __init__(self) -> None:
    """Initialize the AfkController with the afkmodel table."""
    super().__init__("afkmodel")
    self.guild_table: GuildActions[Guild] = db.client.guild

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
Python
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
Python
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
Python
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
Python
    """
    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
Python
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

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

Source code in tux/database/controllers/afk.py
Python
"nickname": nickname,
"reason": reason,
"perm_afk": perm_afk,
"until": until,
_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
Python
        "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
Python
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
Python
    """
    return await self.delete(where={"member_id": member_id})

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
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
Python
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]

Constructs keyword arguments for Prisma create operations.

Source code in tux/database/controllers/afk.py
Python
Returns
-------
int
    The number of AFK members in the guild
"""
return await self.count(where={"guild_id": guild_id})
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
Python
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]

Constructs keyword arguments for Prisma update operations.

Source code in tux/database/controllers/afk.py
Python
"""Get all AFK members in a guild.

Parameters
----------
guild_id : int
    The ID of the guild to get AFK members for

Returns
-------
list[AFKModel]
_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/afk.py
Python
"""
return await self.find_many(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 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

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.

create(data: dict[str, Any], include: dict[str, bool] | None = None) -> AFKModel 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.

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 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.