afk ¶

AFK (Away From Keyboard) status management controller.

This controller manages AFK status for Discord guild members, including temporary and permanent AFK states with customizable messages and time limits.

Classes:

AfkController –

Clean AFK controller using the new BaseController pattern.

Classes¶

AfkController ¶

Python

AfkController(db: DatabaseService | None = None)

Bases: BaseController[AFK]

Clean AFK controller using the new BaseController pattern.

Initialize the AFK controller.

Parameters:

db (DatabaseService | None, default: None ) –

The database service instance. If None, uses the default service.

Returned by:

controllers Classes DatabaseCoordinator Attributes afk

Methods:

get_afk_by_member –

Get AFK status for a specific member in a guild.
set_member_afk –

Set a member as AFK.
remove_member_afk –

Remove AFK status for a member.
get_all_afk_members –

Get all members currently AFK in a guild.
is_member_afk –

Check if a member is AFK in a guild.
is_member_perm_afk –

Check if a member is permanently AFK in a guild.
is_afk –

Check if a member is currently AFK - alias for is_member_afk.
get_afk_member –

Get AFK record for a member - alias for get_afk_by_member.
remove_afk –

Remove AFK status for a member - alias for remove_member_afk.
create –

Create a new record.
set_afk –

Set a member as AFK - alias for set_member_afk.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
find_many –

Find many AFK records with optional filters - alias for find_all.
delete_by_id –

Delete a record by ID.
is_perm_afk –

Check if a member is permanently AFK - alias for is_member_perm_afk.
exists –

Check if a record exists.
get_expired_afk_members –

Get all expired AFK members in a guild.
find_one –

Find one record.
find_all –

Find all records with performance optimizations.
find_all_with_options –

Find all records with relationship loading options.
count –

Count records.
get_all –

Get all records (alias for find_all without pagination).
execute_query –

Execute a custom query.
find_with_json_query –

Find records using JSON column queries.
find_with_array_contains –

Find records where array column contains value.
find_with_full_text_search –

Find records using full-text search.
paginate –

Paginate records with metadata.
find_paginated –

Find paginated records with relationship loading.
bulk_create –

Create multiple records in bulk.
bulk_update –

Update multiple records in bulk.
bulk_delete –

Delete multiple records in bulk.
update_where –

Update records matching filters.
delete_where –

Delete records matching filters.
bulk_upsert_with_conflict_resolution –

Bulk upsert with conflict resolution.
with_session –

Execute operation within a session context.
with_transaction –

Execute operation within a transaction context.
execute_transaction –

Execute a callback within a transaction.
get_table_statistics –

Get comprehensive table statistics.
explain_query_performance –

Explain query performance with optional analysis.
upsert_by_field –

Upsert a record by a specific field.
upsert_by_id –

Upsert a record by ID.
get_or_create_by_field –

Get existing record or create new one by field.
get_or_create –

Get existing record or create new one.
upsert –

Upsert a record.

Attributes:

db_service (DatabaseService) –

Database service property for test compatibility.
model_class (type[ModelT]) –

Model class property for test compatibility.

Attributes¶

db_service `property` ¶

Python

db_service: DatabaseService

Database service property for test compatibility.

Returns:

DatabaseService –

The database service instance.

model_class `property` ¶

Python

model_class: type[ModelT]

Model class property for test compatibility.

Returns:

type[ModelT] –

The SQLModel class.

Functions¶

get_afk_by_member `async` ¶

Python

get_afk_by_member(member_id: int, guild_id: int) -> AFK | None

Get AFK status for a specific member in a guild.

Returns:

AFK | None –

The AFK record if found, None otherwise.

set_member_afk `async` ¶

Python

set_member_afk(
    member_id: int,
    nickname: str,
    reason: str,
    guild_id: int,
    is_perm: bool = False,
    until: datetime | None = None,
    enforced: bool = False,
) -> AFK

Set a member as AFK.

Returns:

AFK –

The AFK record (created or updated).

remove_member_afk `async` ¶

Python

remove_member_afk(member_id: int, guild_id: int) -> bool

Remove AFK status for a member.

Returns:

bool –

True if removed successfully, False otherwise.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

get_all_afk_members `async` ¶

Python

get_all_afk_members(guild_id: int) -> list[AFK]

Get all members currently AFK in a guild.

Returns:

list[AFK] –

List of all AFK records for the guild.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

is_member_afk `async` ¶

Python

is_member_afk(member_id: int, guild_id: int) -> bool

Check if a member is AFK in a guild.

Returns:

bool –

True if member is AFK, False otherwise.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

is_member_perm_afk `async` ¶

Python

is_member_perm_afk(member_id: int, guild_id: int) -> bool

Check if a member is permanently AFK in a guild.

Returns:

bool –

True if member is permanently AFK, False otherwise.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

is_afk `async` ¶

Python

is_afk(member_id: int, guild_id: int) -> bool

Check if a member is currently AFK - alias for is_member_afk.

Returns:

bool –

True if member is AFK, False otherwise.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

get_afk_member `async` ¶

Python

get_afk_member(member_id: int, guild_id: int) -> AFK | None

Get AFK record for a member - alias for get_afk_by_member.

Returns:

AFK | None –

The AFK record if found, None otherwise.

remove_afk `async` ¶

Python

remove_afk(member_id: int, guild_id: int) -> bool

Remove AFK status for a member - alias for remove_member_afk.

Returns:

bool –

True if removed successfully, False otherwise.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

set_afk `async` ¶

Python

set_afk(
    member_id: int,
    nickname: str,
    reason: str,
    guild_id: int,
    is_perm: bool,
    until: datetime | None = None,
    enforced: bool = False,
) -> AFK

Set a member as AFK - alias for set_member_afk.

Returns:

AFK –

The AFK record (created or updated).

get_by_id `async` ¶

Python

get_by_id(record_id: Any) -> ModelT | None

Get a record by ID.

Returns:

ModelT | None –

The record if found, None otherwise.

update_by_id `async` ¶

Python

update_by_id(record_id: Any, **values: Any) -> ModelT | None

Update a record by ID.

Returns:

ModelT | None –

The updated record, or None if not found.

find_many `async` ¶

Python

find_many(**filters: Any) -> list[AFK]

Find many AFK records with optional filters - alias for find_all.

Returns:

list[AFK] –

List of AFK records matching the filters.

delete_by_id `async` ¶

Python

delete_by_id(record_id: Any) -> bool

Delete a record by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

is_perm_afk `async` ¶

Python

is_perm_afk(member_id: int, guild_id: int) -> bool

Check if a member is permanently AFK - alias for is_member_perm_afk.

Returns:

bool –

True if member is permanently AFK, False otherwise.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

get_expired_afk_members `async` ¶

Python

get_expired_afk_members(guild_id: int) -> list[AFK]

Get all expired AFK members in a guild.

Returns:

list[AFK] –

List of expired AFK records (currently returns empty list).

find_one `async` ¶

Python

find_one(filters: Any | None = None, order_by: Any | None = None) -> ModelT | None

Find one record.

Returns:

ModelT | None –

The found record, or None if not found.

find_all `async` ¶

Python

find_all(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
) -> list[ModelT]

Find all records with performance optimizations.

Returns:

list[ModelT] –

List of found records.

find_all_with_options `async` ¶

Python

find_all_with_options(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
    load_relationships: list[str] | None = None,
) -> list[ModelT]

Find all records with relationship loading options.

Returns:

list[ModelT] –

List of found records with loaded relationships.

count `async` ¶

Python

count(filters: Any | None = None) -> int

Count records.

Returns:

int –

The count of matching records.

get_all `async` ¶

Python

get_all(filters: Any | None = None, order_by: Any | None = None) -> list[ModelT]

Get all records (alias for find_all without pagination).

Returns:

list[ModelT] –

List of all matching records.

execute_query `async` ¶

Python

execute_query(query: Any) -> Any

Execute a custom query.

Returns:

Any –

The query result.

find_with_json_query `async` ¶

Python

find_with_json_query(
    json_column: str, json_path: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records using JSON column queries.

Returns:

list[ModelT] –

List of records matching the JSON query.

find_with_array_contains `async` ¶

Python

find_with_array_contains(
    array_column: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records where array column contains value.

Returns:

list[ModelT] –

List of records with matching array values.

find_with_full_text_search `async` ¶

Python

find_with_full_text_search(
    search_columns: list[str], search_term: str, filters: Any | None = None
) -> list[ModelT]

Find records using full-text search.

Returns:

list[ModelT] –

List of records matching the search term.

paginate `async` ¶

Python

paginate(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
) -> PaginationResult[ModelT]

Paginate records with metadata.

Returns:

PaginationResult[ModelT] –

Pagination result with items, total, and page info.

find_paginated `async` ¶

Python

find_paginated(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
    load_relationships: list[str] | None = None,
) -> PaginationResult[ModelT]

Find paginated records with relationship loading.

Returns:

PaginationResult[ModelT] –

Pagination result with items and relationships loaded.

bulk_create `async` ¶

Python

bulk_create(items: list[dict[str, Any]]) -> list[ModelT]

Create multiple records in bulk.

Returns:

list[ModelT] –

List of created records.

bulk_update `async` ¶

Python

bulk_update(updates: list[tuple[Any, dict[str, Any]]]) -> int

Update multiple records in bulk.

Returns:

int –

Number of records updated.

bulk_delete `async` ¶

Python

bulk_delete(record_ids: list[Any]) -> int

Delete multiple records in bulk.

Returns:

int –

Number of records deleted.

update_where `async` ¶

Python

update_where(filters: Any, values: dict[str, Any]) -> int

Update records matching filters.

Returns:

int –

Number of records updated.

delete_where `async` ¶

Python

delete_where(filters: Any) -> int

Delete records matching filters.

Returns:

int –

Number of records deleted.

bulk_upsert_with_conflict_resolution `async` ¶

Python

bulk_upsert_with_conflict_resolution(
    items: list[dict[str, Any]],
    conflict_columns: list[str],
    update_columns: list[str] | None = None,
) -> list[ModelT]

Bulk upsert with conflict resolution.

Returns:

list[ModelT] –

List of upserted records.

with_session `async` ¶

Python

with_session[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a session context.

Returns:

R –

The result of the operation.

with_transaction `async` ¶

Python

with_transaction[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a transaction context.

Returns:

R –

The result of the operation.

execute_transaction `async` ¶

Python

execute_transaction(callback: Callable[[], Any]) -> Any

Execute a callback within a transaction.

Returns:

Any –

The result of the callback.

get_table_statistics `async` ¶

Python

get_table_statistics() -> dict[str, Any]

Get comprehensive table statistics.

Returns:

dict[str, Any] –

Dictionary containing table statistics.

explain_query_performance `async` ¶

Python

explain_query_performance(
    query: Any, analyze: bool = False, buffers: bool = False
) -> dict[str, Any]

Explain query performance with optional analysis.

Returns:

dict[str, Any] –

Dictionary containing query execution plan and statistics.

upsert_by_field `async` ¶

Python

upsert_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Upsert a record by a specific field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert_by_id `async` ¶

Python

upsert_by_id(
    record_id: Any, defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record by ID.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create_by_field `async` ¶

Python

get_or_create_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Get existing record or create new one by field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create `async` ¶

Python

get_or_create(
    defaults: dict[str, Any] | None = None, **filters: Any
) -> tuple[ModelT, bool]

Get existing record or create new one.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert `async` ¶

Python

upsert(
    filters: dict[str, Any], defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

afk ¶

Classes¶

AfkController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

get_afk_by_member async ¶

set_member_afk async ¶

remove_member_afk async ¶

_get_pagination ¶

get_all_afk_members async ¶

_get_bulk ¶

is_member_afk async ¶

_get_transaction ¶

is_member_perm_afk async ¶

_get_performance ¶

is_afk async ¶

_get_upsert ¶

get_afk_member async ¶

remove_afk async ¶

create async ¶

set_afk async ¶

get_by_id async ¶

update_by_id async ¶

find_many async ¶

delete_by_id async ¶

is_perm_afk async ¶

exists async ¶

get_expired_afk_members async ¶

find_one async ¶

find_all async ¶

find_all_with_options async ¶

count async ¶

get_all async ¶

execute_query async ¶

find_with_json_query async ¶

find_with_array_contains async ¶

find_with_full_text_search async ¶

paginate async ¶

find_paginated async ¶

bulk_create async ¶

bulk_update async ¶

bulk_delete async ¶

update_where async ¶

delete_where async ¶

bulk_upsert_with_conflict_resolution async ¶

with_session async ¶

with_transaction async ¶

execute_transaction async ¶

get_table_statistics async ¶

explain_query_performance async ¶

upsert_by_field async ¶

upsert_by_id async ¶

get_or_create_by_field async ¶

get_or_create async ¶

upsert async ¶

db_service `property` ¶

model_class `property` ¶

get_afk_by_member `async` ¶

set_member_afk `async` ¶

remove_member_afk `async` ¶

get_all_afk_members `async` ¶

is_member_afk `async` ¶

is_member_perm_afk `async` ¶

is_afk `async` ¶

get_afk_member `async` ¶

remove_afk `async` ¶

create `async` ¶

set_afk `async` ¶

get_by_id `async` ¶

update_by_id `async` ¶

find_many `async` ¶

delete_by_id `async` ¶

is_perm_afk `async` ¶

exists `async` ¶

get_expired_afk_members `async` ¶

find_one `async` ¶

find_all `async` ¶

find_all_with_options `async` ¶

count `async` ¶

get_all `async` ¶

execute_query `async` ¶

find_with_json_query `async` ¶

find_with_array_contains `async` ¶

find_with_full_text_search `async` ¶

paginate `async` ¶

find_paginated `async` ¶

bulk_create `async` ¶

bulk_update `async` ¶

bulk_delete `async` ¶

update_where `async` ¶

delete_where `async` ¶

bulk_upsert_with_conflict_resolution `async` ¶

with_session `async` ¶

with_transaction `async` ¶

execute_transaction `async` ¶

get_table_statistics `async` ¶

explain_query_performance `async` ¶

upsert_by_field `async` ¶

upsert_by_id `async` ¶

get_or_create_by_field `async` ¶

get_or_create `async` ¶

upsert `async` ¶