controllers ¶

Database Controllers for Tux Bot.

This module provides the controller layer for database operations, offering lazy-loaded controllers for different data models and coordinated access to database functionality.

Modules:

afk –

AFK (Away From Keyboard) status management controller.
base –

Database controller components for modular database operations.
case –

Moderation case management controller.
guild –

Guild and guild configuration management controller.
guild_config –

Guild configuration management controller.
levels –

User level and XP tracking controller.
permissions –

Dynamic permission system controllers.
reminder –

User reminder scheduling controller.
snippet –

Code snippet storage and management controller.
starboard –

Starboard message highlighting controller.

Classes:

AfkController –

Clean AFK controller using the new BaseController pattern.
BaseController –

Composed database controller with lazy-loaded specialized operations.
CaseController –

Clean Case controller using the new BaseController pattern.
GuildController –

Clean Guild controller using the new BaseController pattern.
GuildConfigController –

Clean GuildConfig controller using the new BaseController pattern.
LevelsController –

Clean Levels controller using the new BaseController pattern.
PermissionAssignmentController –

Controller for managing guild permission assignments.
PermissionCommandController –

Controller for managing command permission requirements.
PermissionRankController –

Controller for managing guild permission ranks.
ReminderController –

Clean Reminder controller using the new BaseController pattern.
SnippetController –

Clean Snippet controller using the new BaseController pattern.
StarboardController –

Clean Starboard controller using the new BaseController pattern.
StarboardMessageController –

Clean StarboardMessage controller using the new BaseController pattern.
DatabaseCoordinator –

Coordinator for database controllers with lazy loading.

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

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

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

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

Get all expired AFK members in a guild.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_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.

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.

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.

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

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

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.

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.

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

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

BaseController ¶

Python

BaseController(model: type[ModelT], db: DatabaseService | None = None)

Composed database controller with lazy-loaded specialized operations.

This controller delegates operations to specialized controllers while maintaining backward compatibility with the original BaseController API. Core CRUD and Query controllers are eagerly initialized, while specialized controllers (pagination, bulk, transaction, performance, upsert) use lazy initialization to reduce overhead for simple use cases.

Initialize the base controller with lazy-loaded specialized controllers.

Parameters:

model (type[ModelT]) –

The SQLModel class to perform operations on.
db (DatabaseService | None, default: None ) –

The database service instance. Must be provided.

Raises:

RuntimeError –

If db is None, as database service is required.

Returned by:

utils Functions create_enhanced_controller_from

Subclassed by:

Methods:

create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

CaseController ¶

Python

CaseController(db: DatabaseService | None = None)

Bases: BaseController[Case]

Clean Case controller using the new BaseController pattern.

Initialize the case controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes case

Used by:

Methods:

get_case_by_id –

Get a case by its ID.
get_cases_by_user –

Get all cases for a specific user in a guild.
get_active_cases_by_user –

Get all active cases for a specific user in a guild.
create_case –

Create a new case with auto-generated case number.
update_case –

Update a case by ID.
update_mod_log_message_id –

Update the mod log message ID for a case.
close_case –

Close a case by setting its status to False.
delete_case –

Delete a case by ID.
get_cases_by_guild –

Get all cases for a guild, optionally limited.
get_cases_by_type –

Get all cases of a specific type in a guild.
get_recent_cases –

Get cases created within the last N hours.
get_case_count_by_guild –

Get the total number of cases in a guild.
is_user_under_restriction –

Check if a user is under any active restriction in a guild.
get_case_by_number –

Get a case by its case number in a guild.
get_cases_by_options –

Get cases by various filter options.
update_case_by_number –

Update a case by guild ID and case number.
get_all_cases –

Get all cases in a guild.
get_latest_case_by_user –

Get the most recent case for a user in a guild.
set_tempban_expired –

Mark a tempban case as processed after the user has been unbanned.
get_expired_tempbans –

Get tempban cases that have expired but haven't been processed yet.
get_case_count_by_user –

Get the total number of cases for a specific user in a guild.
get_cases_by_moderator –

Get all cases moderated by a specific user in a guild.
get_expired_cases –

Get all expired cases (any type) that haven't been processed yet.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_case_by_id `async` ¶

Python

get_case_by_id(case_id: int) -> Case | None

Get a case by its ID.

Returns:

Case | None –

The case if found, None otherwise.

get_cases_by_user `async` ¶

Python

get_cases_by_user(user_id: int, guild_id: int) -> list[Case]

Get all cases for a specific user in a guild.

Returns:

list[Case] –

List of all cases for the user in the guild.

get_active_cases_by_user `async` ¶

Python

get_active_cases_by_user(user_id: int, guild_id: int) -> list[Case]

Get all active cases for a specific user in a guild.

Returns:

list[Case] –

List of active cases for the user in the guild.

create_case `async` ¶

Python

create_case(
    case_type: str,
    case_user_id: int,
    case_moderator_id: int,
    guild_id: int,
    case_reason: str | None = None,
    case_status: bool = True,
    **kwargs: Any,
) -> Case

Create a new case with auto-generated case number.

Uses SELECT FOR UPDATE to prevent race conditions when generating case numbers.

Parameters:

case_type (str) –

The type of case (from CaseType enum value)
case_user_id (int) –

Discord ID of the user being moderated
case_moderator_id (int) –

Discord ID of the moderator
guild_id (int) –

Discord guild ID
case_reason (str | None, default: None ) –

Reason for the moderation action
case_status (bool, default: True ) –

Whether the case is active (default True)
**kwargs (Any, default: {} ) –

Additional case fields (e.g., case_expires_at, case_metadata, mod_log_message_id)

Returns:

Case –

The newly created case with auto-generated case number.

Notes

For expiring cases, use case_expires_at (datetime) in kwargs
Do NOT pass duration - convert to case_expires_at before calling this method
Case numbers are auto-generated per guild using SELECT FOR UPDATE locking

update_case `async` ¶

Python

update_case(case_id: int, **kwargs: Any) -> Case | None

Update a case by ID.

Returns:

Case | None –

The updated case, or None if not found.

update_mod_log_message_id `async` ¶

Python

update_mod_log_message_id(case_id: int, message_id: int) -> Case | None

Update the mod log message ID for a case.

Parameters:

case_id (int) –

The case ID to update.
message_id (int) –

The Discord message ID from the mod log.

Returns:

Case | None –

The updated case, or None if not found.

close_case `async` ¶

Python

close_case(case_id: int) -> Case | None

Close a case by setting its status to False.

Returns:

Case | None –

The updated case, or None if not found.

delete_case `async` ¶

Python

delete_case(case_id: int) -> bool

Delete a case by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

get_cases_by_guild `async` ¶

Python

get_cases_by_guild(guild_id: int, limit: int | None = None) -> list[Case]

Get all cases for a guild, optionally limited.

Returns:

list[Case] –

List of cases for the guild.

get_cases_by_type `async` ¶

Python

get_cases_by_type(guild_id: int, case_type: str) -> list[Case]

Get all cases of a specific type in a guild.

Returns:

list[Case] –

List of cases matching the specified type.

get_recent_cases `async` ¶

Python

get_recent_cases(guild_id: int, hours: int = 24) -> list[Case]

Get cases created within the last N hours.

Returns:

list[Case] –

List of recent cases.

get_case_count_by_guild `async` ¶

Python

get_case_count_by_guild(guild_id: int) -> int

Get the total number of cases in a guild.

Returns:

int –

The total count of cases in the guild.

is_user_under_restriction `async` ¶

Python

is_user_under_restriction(
    user_id: int | None = None, guild_id: int | None = None, **kwargs: Any
) -> bool

Check if a user is under any active restriction in a guild.

Returns:

bool –

True if user is under restriction, False otherwise.

get_case_by_number `async` ¶

Python

get_case_by_number(case_number: int, guild_id: int) -> Case | None

Get a case by its case number in a guild.

Returns:

Case | None –

The case if found, None otherwise.

get_cases_by_options `async` ¶

Python

get_cases_by_options(
    guild_id: int, options: dict[str, Any] | None = None
) -> list[Case]

Get cases by various filter options.

Returns:

list[Case] –

List of cases matching the specified options.

update_case_by_number `async` ¶

Python

update_case_by_number(guild_id: int, case_number: int, **kwargs: Any) -> Case | None

Update a case by guild ID and case number.

Returns:

Case | None –

The updated case, or None if not found.

get_all_cases `async` ¶

Python

get_all_cases(guild_id: int) -> list[Case]

Get all cases in a guild.

Returns:

list[Case] –

List of all cases in the guild.

get_latest_case_by_user `async` ¶

Python

get_latest_case_by_user(user_id: int, guild_id: int) -> Case | None

Get the most recent case for a user in a guild.

Returns:

Case | None –

The most recent case if found, None otherwise.

set_tempban_expired `async` ¶

Python

set_tempban_expired(case_id: int, guild_id: int | None = None) -> bool

Mark a tempban case as processed after the user has been unbanned.

This sets case_processed=True to indicate the expiration has been handled. The case_status remains True (the case is still valid, just completed). The case_expires_at field remains unchanged as a historical record.

Parameters:

case_id (int) –

The ID of the case to mark as processed
guild_id (int | None, default: None ) –

Deprecated parameter kept for backward compatibility (unused)

Returns:

bool –

True if the case was updated, False if not found

get_expired_tempbans `async` ¶

Python

get_expired_tempbans(guild_id: int) -> list[Case]

Get tempban cases that have expired but haven't been processed yet.

Returns:

list[Case] –

List of expired unprocessed tempban cases where case_expires_at is in the past, case_processed=False, and case_status=True.

get_case_count_by_user `async` ¶

Python

get_case_count_by_user(user_id: int, guild_id: int) -> int

Get the total number of cases for a specific user in a guild.

Returns:

int –

The total count of cases for the user.

get_cases_by_moderator `async` ¶

Python

get_cases_by_moderator(moderator_id: int, guild_id: int) -> list[Case]

Get all cases moderated by a specific user in a guild.

Returns:

list[Case] –

List of cases moderated by the user.

get_expired_cases `async` ¶

Python

get_expired_cases(guild_id: int) -> list[Case]

Get all expired cases (any type) that haven't been processed yet.

Returns:

list[Case] –

List of expired unprocessed cases where case_expires_at is in the past, case_processed=False, and case_status=True.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

GuildController ¶

Python

GuildController(db: DatabaseService | None = None)

Bases: BaseController[Guild]

Clean Guild controller using the new BaseController pattern.

Initialize the guild controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes guild

Methods:

get_guild_by_id –

Get a guild by its ID.
get_or_create_guild –

Get a guild by ID, or create it if it doesn't exist.
create_guild –

Create a new guild.
delete_guild –

Delete a guild by ID.
get_guild_config –

Get guild configuration.
update_guild_config –

Update guild configuration.
get_all_guilds –

Get all guilds.
get_guild_count –

Get the total number of guilds.
find_many –

Find many guilds with optional filters - alias for find_all.
insert_guild_by_id –

Insert a new guild by ID.
delete_guild_by_id –

Delete a guild by ID.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_guild_by_id `async` ¶

Python

get_guild_by_id(guild_id: int) -> Guild | None

Get a guild by its ID.

Returns:

Guild | None –

The guild if found, None otherwise.

get_or_create_guild `async` ¶

Python

get_or_create_guild(guild_id: int) -> Guild

Get a guild by ID, or create it if it doesn't exist.

Returns:

Guild –

The guild (existing or newly created).

create_guild `async` ¶

Python

create_guild(guild_id: int) -> Guild

Create a new guild.

Returns:

Guild –

The newly created guild.

delete_guild `async` ¶

Python

delete_guild(guild_id: int) -> bool

Delete a guild by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

get_guild_config `async` ¶

Python

get_guild_config(guild_id: int) -> GuildConfig | None

Get guild configuration.

Returns:

GuildConfig | None –

The guild configuration if found, None otherwise.

update_guild_config `async` ¶

Python

update_guild_config(guild_id: int, data: dict[str, Any]) -> GuildConfig

Update guild configuration.

Returns:

GuildConfig –

The updated guild configuration.

get_all_guilds `async` ¶

Python

get_all_guilds() -> list[Guild]

Get all guilds.

Returns:

list[Guild] –

List of all guilds.

get_guild_count `async` ¶

Python

get_guild_count() -> int

Get the total number of guilds.

Returns:

int –

The total count of guilds.

find_many `async` ¶

Python

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

Find many guilds with optional filters - alias for find_all.

Returns:

list[Guild] –

List of guilds matching the filters.

insert_guild_by_id `async` ¶

Python

insert_guild_by_id(guild_id: int, **kwargs: Any) -> Guild

Insert a new guild by ID.

Returns:

Guild –

The newly created guild.

delete_guild_by_id `async` ¶

Python

delete_guild_by_id(guild_id: int) -> bool

Delete a guild by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

GuildConfigController ¶

Python

GuildConfigController(db: DatabaseService | None = None)

Bases: BaseController[GuildConfig]

Clean GuildConfig controller using the new BaseController pattern.

Initialize the guild config controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes guild_config

Methods:

get_config_by_guild_id –

Get guild configuration by guild ID.
get_or_create_config –

Get guild configuration, or create it with defaults if it doesn't exist.
update_config –

Update guild configuration.
delete_config –

Delete guild configuration.
get_all_configs –

Get all guild configurations.
get_config_count –

Get the total number of guild configurations.
find_configs_by_field –

Find configurations by a specific field value.
update_config_field –

Update a specific field in guild configuration.
update_onboarding_stage –

Update the onboarding stage for a guild.
mark_onboarding_completed –

Mark onboarding as completed for a guild.
reset_onboarding –

Reset onboarding status for a guild.
get_onboarding_status –

Get onboarding status for a guild.
update_channel_field –

Update a channel field in guild configuration.
get_configs_by_prefix –

Get configurations where guild ID starts with a prefix.
update_perm_level_role –

Update permission level role for a guild.
get_config_field –

Get any field from guild configuration.
get_jail_role_id –

Get jail role ID for a guild.
get_perm_level_role –

Get role ID for a specific permission level.
get_jail_channel_id –

Get jail channel ID for a guild.
update_private_log_id –

Update private log channel ID.
update_report_log_id –

Update report log channel ID.
update_dev_log_id –

Update dev log channel ID.
update_mod_log_id –

Update mod log channel ID.
update_audit_log_id –

Update audit log channel ID.
update_join_log_id –

Update join log channel ID.
update_jail_channel_id –

Update jail channel ID.
update_starboard_channel_id –

Update starboard channel ID.
update_general_channel_id –

Update general channel ID.
get_starboard_channel_id –

Get starboard channel ID for a guild.
get_general_channel_id –

Get general channel ID for a guild.
get_join_log_id –

Get join log channel ID for a guild.
get_audit_log_id –

Get audit log channel ID for a guild.
get_mod_log_id –

Get mod log channel ID for a guild.
get_private_log_id –

Get private log channel ID for a guild.
get_report_log_id –

Get report log channel ID for a guild.
get_dev_log_id –

Get dev log channel ID for a guild.
update_guild_prefix –

Update guild prefix.
delete_guild_prefix –

Delete guild prefix (set to default).
get_log_channel –

Get log channel ID for a guild based on log type.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_config_by_guild_id `async` ¶

Python

get_config_by_guild_id(guild_id: int) -> GuildConfig | None

Get guild configuration by guild ID.

Returns:

GuildConfig | None –

The guild configuration if found, None otherwise.

get_or_create_config `async` ¶

Python

get_or_create_config(guild_id: int, **defaults: Any) -> GuildConfig

Get guild configuration, or create it with defaults if it doesn't exist.

Returns:

GuildConfig –

The guild configuration (existing or newly created).

update_config `async` ¶

Python

update_config(guild_id: int, **updates: Any) -> GuildConfig | None

Update guild configuration.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

delete_config `async` ¶

Python

delete_config(guild_id: int) -> bool

Delete guild configuration.

Returns:

bool –

True if deleted successfully, False otherwise.

get_all_configs `async` ¶

Python

get_all_configs() -> list[GuildConfig]

Get all guild configurations.

Returns:

list[GuildConfig] –

List of all guild configurations.

get_config_count `async` ¶

Python

get_config_count() -> int

Get the total number of guild configurations.

Returns:

int –

The total count of guild configurations.

find_configs_by_field `async` ¶

Python

find_configs_by_field(field_name: str, field_value: Any) -> list[GuildConfig]

Find configurations by a specific field value.

Returns:

list[GuildConfig] –

List of configurations with matching field value.

update_config_field `async` ¶

Python

update_config_field(
    guild_id: int, field_name: str, field_value: Any
) -> GuildConfig | None

Update a specific field in guild configuration.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_onboarding_stage `async` ¶

Python

update_onboarding_stage(guild_id: int, stage: str) -> GuildConfig | None

Update the onboarding stage for a guild.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

mark_onboarding_completed `async` ¶

Python

mark_onboarding_completed(guild_id: int) -> GuildConfig | None

Mark onboarding as completed for a guild.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

reset_onboarding `async` ¶

Python

reset_onboarding(guild_id: int) -> GuildConfig | None

Reset onboarding status for a guild.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

get_onboarding_status `async` ¶

Python

get_onboarding_status(guild_id: int) -> tuple[bool, str | None]

Get onboarding status for a guild.

Returns:

tuple[bool, str | None] –

Tuple of (completed, stage) for the guild's onboarding status.

update_channel_field `async` ¶

Python

update_channel_field(
    guild_id: int, channel_field: str, channel_id: int
) -> GuildConfig | None

Update a channel field in guild configuration.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

get_configs_by_prefix `async` ¶

Python

get_configs_by_prefix(prefix: str) -> list[GuildConfig]

Get configurations where guild ID starts with a prefix.

Returns:

list[GuildConfig] –

List of configurations with matching guild ID prefix.

update_perm_level_role `async` ¶

Python

update_perm_level_role(
    guild_id: int, role_id: int | str, perm_level: int | str
) -> GuildConfig | None

Update permission level role for a guild.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

get_config_field `async` ¶

Python

get_config_field(guild_id: int, field_name: str) -> Any

Get any field from guild configuration.

Returns:

Any –

The field value, or None if configuration or field not found.

get_jail_role_id `async` ¶

Python

get_jail_role_id(guild_id: int) -> int | None

Get jail role ID for a guild.

Returns:

int | None –

The jail role ID, or None if not configured.

get_perm_level_role `async` ¶

Python

get_perm_level_role(guild_id: int, perm_level: str) -> int | None

Get role ID for a specific permission level.

Returns:

int | None –

The role ID for the permission level, or None if not configured.

get_jail_channel_id `async` ¶

Python

get_jail_channel_id(guild_id: int) -> int | None

Get jail channel ID for a guild.

Returns:

int | None –

The jail channel ID, or None if not configured.

update_private_log_id `async` ¶

Python

update_private_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update private log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_report_log_id `async` ¶

Python

update_report_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update report log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_dev_log_id `async` ¶

Python

update_dev_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update dev log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_mod_log_id `async` ¶

Python

update_mod_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update mod log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_audit_log_id `async` ¶

Python

update_audit_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update audit log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_join_log_id `async` ¶

Python

update_join_log_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update join log channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_jail_channel_id `async` ¶

Python

update_jail_channel_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update jail channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_starboard_channel_id `async` ¶

Python

update_starboard_channel_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update starboard channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

update_general_channel_id `async` ¶

Python

update_general_channel_id(guild_id: int, channel_id: int) -> GuildConfig | None

Update general channel ID.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

get_starboard_channel_id `async` ¶

Python

get_starboard_channel_id(guild_id: int) -> int | None

Get starboard channel ID for a guild.

Returns:

int | None –

The starboard channel ID, or None if not configured.

get_general_channel_id `async` ¶

Python

get_general_channel_id(guild_id: int) -> int | None

Get general channel ID for a guild.

Returns:

int | None –

The general channel ID, or None if not configured.

get_join_log_id `async` ¶

Python

get_join_log_id(guild_id: int) -> int | None

Get join log channel ID for a guild.

Returns:

int | None –

The join log channel ID, or None if not configured.

get_audit_log_id `async` ¶

Python

get_audit_log_id(guild_id: int) -> int | None

Get audit log channel ID for a guild.

Returns:

int | None –

The audit log channel ID, or None if not configured.

get_mod_log_id `async` ¶

Python

get_mod_log_id(guild_id: int) -> int | None

Get mod log channel ID for a guild.

Returns:

int | None –

The mod log channel ID, or None if not configured.

get_private_log_id `async` ¶

Python

get_private_log_id(guild_id: int) -> int | None

Get private log channel ID for a guild.

Returns:

int | None –

The private log channel ID, or None if not configured.

get_report_log_id `async` ¶

Python

get_report_log_id(guild_id: int) -> int | None

Get report log channel ID for a guild.

Returns:

int | None –

The report log channel ID, or None if not configured.

get_dev_log_id `async` ¶

Python

get_dev_log_id(guild_id: int) -> int | None

Get dev log channel ID for a guild.

Returns:

int | None –

The dev log channel ID, or None if not configured.

update_guild_prefix `async` ¶

Python

update_guild_prefix(guild_id: int, prefix: str) -> GuildConfig | None

Update guild prefix.

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

delete_guild_prefix `async` ¶

Python

delete_guild_prefix(guild_id: int) -> GuildConfig | None

Delete guild prefix (set to default).

Returns:

GuildConfig | None –

The updated configuration, or None if not found.

get_log_channel `async` ¶

Python

get_log_channel(guild_id: int, log_type: str | None = None) -> int | None

Get log channel ID for a guild based on log type.

Returns:

int | None –

The log channel ID for the specified type, or None if not found.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

LevelsController ¶

Python

LevelsController(db: DatabaseService | None = None)

Bases: BaseController[Levels]

Clean Levels controller using the new BaseController pattern.

Initialize the levels controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes levels

Methods:

get_levels_by_member –

Get levels for a specific member in a guild.
get_or_create_levels –

Get levels for a member, or create them if they don't exist.
add_xp –

Add XP to a member's levels.
set_xp –

Set a member's XP to a specific value.
set_level –

Set a member's level to a specific value.
blacklist_member –

Blacklist a member from gaining XP.
unblacklist_member –

Remove a member from the blacklist.
get_top_members –

Get top members by XP in a guild.
get_xp –

Get XP for a specific member in a guild.
get_level –

Get level for a specific member in a guild.
update_xp_and_level –

Update XP and level for a member.
reset_xp –

Reset XP and level for a member.
toggle_blacklist –

Toggle blacklist status for a member.
is_blacklisted –

Check if a member is blacklisted.
get_last_message_time –

Get the last message time for a member.
get_xp_and_level –

Get both XP and level for a member.
get_member_rank –

Get a member's rank in their guild (1-based).
get_guild_stats –

Get guild statistics.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_levels_by_member `async` ¶

Python

get_levels_by_member(member_id: int, guild_id: int) -> Levels | None

Get levels for a specific member in a guild.

Returns:

Levels | None –

The levels record if found, None otherwise.

get_or_create_levels `async` ¶

Python

get_or_create_levels(member_id: int, guild_id: int) -> Levels

Get levels for a member, or create them if they don't exist.

Returns:

Levels –

The levels record for the member.

add_xp `async` ¶

Python

add_xp(member_id: int, guild_id: int, xp_amount: float) -> Levels

Add XP to a member's levels.

Returns:

Levels –

The updated levels record.

set_xp `async` ¶

Python

set_xp(member_id: int, guild_id: int, xp: float) -> Levels

Set a member's XP to a specific value.

Returns:

Levels –

The updated levels record.

set_level `async` ¶

Python

set_level(member_id: int, guild_id: int, level: int) -> Levels

Set a member's level to a specific value.

Returns:

Levels –

The updated levels record.

blacklist_member `async` ¶

Python

blacklist_member(member_id: int, guild_id: int) -> Levels

Blacklist a member from gaining XP.

Returns:

Levels –

The updated levels record.

unblacklist_member `async` ¶

Python

unblacklist_member(member_id: int, guild_id: int) -> Levels

Remove a member from the blacklist.

Returns:

Levels –

The updated levels record.

get_top_members `async` ¶

Python

get_top_members(guild_id: int, limit: int = 10) -> list[Levels]

Get top members by XP in a guild.

Returns:

list[Levels] –

List of top members sorted by XP (highest first).

get_xp `async` ¶

Python

get_xp(member_id: int, guild_id: int) -> float

Get XP for a specific member in a guild.

Returns:

float –

The member's XP value.

get_level `async` ¶

Python

get_level(member_id: int, guild_id: int) -> int

Get level for a specific member in a guild.

Returns:

int –

The member's level.

update_xp_and_level `async` ¶

Python

update_xp_and_level(
    member_id: int,
    guild_id: int,
    xp_amount: float | None = None,
    new_level: int | None = None,
    last_message: datetime | None = None,
    **kwargs: Any,
) -> Levels

Update XP and level for a member.

Returns:

Levels –

The updated levels record.

Raises:

ValueError –

If xp_amount, new_level, or last_message is missing.

reset_xp `async` ¶

Python

reset_xp(member_id: int, guild_id: int) -> Levels

Reset XP and level for a member.

Returns:

Levels –

The updated levels record with XP and level reset to 0.

toggle_blacklist `async` ¶

Python

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

Toggle blacklist status for a member.

Returns:

bool –

The new blacklist status.

is_blacklisted `async` ¶

Python

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

Check if a member is blacklisted.

Returns:

bool –

True if member is blacklisted, False otherwise.

get_last_message_time `async` ¶

Python

get_last_message_time(member_id: int, guild_id: int) -> datetime

Get the last message time for a member.

Returns:

datetime –

The timestamp of the member's last message.

get_xp_and_level `async` ¶

Python

get_xp_and_level(member_id: int, guild_id: int) -> tuple[float, int]

Get both XP and level for a member.

Returns:

tuple[float, int] –

A tuple containing (xp, level).

get_member_rank `async` ¶

Python

get_member_rank(member_id: int, guild_id: int) -> int

Get a member's rank in their guild (1-based).

Returns:

int –

The member's rank (1 = highest XP), or -1 if blacklisted/not found.

get_guild_stats `async` ¶

Python

get_guild_stats(guild_id: int) -> dict[str, Any]

Get guild statistics.

Returns:

dict[str, Any] –

Dictionary containing total_members, blacklisted_count, and active_members.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

PermissionAssignmentController ¶

Python

PermissionAssignmentController(db: DatabaseService | None = None)

Bases: BaseController[PermissionAssignment]

Controller for managing guild permission assignments.

Initialize the guild permission assignment controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes permission_assignments

Methods:

assign_permission_rank –

Assign a permission level to a role.
get_assignments_by_guild –

Get all permission assignments for a guild.
remove_role_assignment –

Remove a permission level assignment from a role.
get_user_permission_rank –

Get the highest permission rank a user has based on their roles.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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¶

assign_permission_rank `async` ¶

Python

assign_permission_rank(
    guild_id: int, permission_rank_id: int, role_id: int
) -> PermissionAssignment

Assign a permission level to a role.

Returns:

PermissionAssignment –

The newly created permission assignment.

get_assignments_by_guild `async` ¶

Python

get_assignments_by_guild(guild_id: int) -> list[PermissionAssignment]

Get all permission assignments for a guild.

Returns:

list[PermissionAssignment] –

List of all permission assignments for the guild.

remove_role_assignment `async` ¶

Python

remove_role_assignment(guild_id: int, role_id: int) -> bool

Remove a permission level assignment from a role.

Returns:

bool –

True if removed successfully, False otherwise.

get_user_permission_rank `async` ¶

Python

get_user_permission_rank(guild_id: int, user_id: int, user_roles: list[int]) -> int

Get the highest permission rank a user has based on their roles.

Returns:

int –

The highest permission rank (0 if user has no assigned roles).

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

PermissionCommandController ¶

Python

PermissionCommandController(db: DatabaseService | None = None)

Bases: BaseController[PermissionCommand]

Controller for managing command permission requirements.

Initialize the guild command permission controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes command_permissions

Methods:

set_command_permission –

Set the permission rank required for a command.
get_command_permission –

Get the permission requirement for a specific command.
get_all_command_permissions –

Get all command permissions for a guild.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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¶

set_command_permission `async` ¶

Python

set_command_permission(
    guild_id: int, command_name: str, required_rank: int, description: str | None = None
) -> PermissionCommand

Set the permission rank required for a command.

Returns:

PermissionCommand –

The command permission record (created or updated).

get_command_permission `async` ¶

Python

get_command_permission(guild_id: int, command_name: str) -> PermissionCommand | None

Get the permission requirement for a specific command.

Returns:

PermissionCommand | None –

The command permission record if found, None otherwise.

get_all_command_permissions `async` ¶

Python

get_all_command_permissions(guild_id: int) -> list[PermissionCommand]

Get all command permissions for a guild.

Returns:

list[PermissionCommand] –

List of all command permissions ordered by name.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

PermissionRankController ¶

Python

PermissionRankController(db: DatabaseService | None = None)

Bases: BaseController[PermissionRank]

Controller for managing guild permission ranks.

Initialize the guild permission rank controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes permission_ranks

Methods:

create_permission_rank –

Create a new permission rank for a guild.
get_permission_ranks_by_guild –

Get all permission ranks for a guild.
get_permission_rank –

Get a specific permission rank.
update_permission_rank –

Update a permission rank.
delete_permission_rank –

Delete a permission rank.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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¶

create_permission_rank `async` ¶

Python

create_permission_rank(
    guild_id: int, rank: int, name: str, description: str | None = None
) -> PermissionRank

Create a new permission rank for a guild.

Returns:

PermissionRank –

The newly created permission rank.

get_permission_ranks_by_guild `async` ¶

Python

get_permission_ranks_by_guild(guild_id: int) -> list[PermissionRank]

Get all permission ranks for a guild.

Returns:

list[PermissionRank] –

List of permission ranks ordered by rank value.

get_permission_rank `async` ¶

Python

get_permission_rank(guild_id: int, rank: int) -> PermissionRank | None

Get a specific permission rank.

Returns:

PermissionRank | None –

The permission rank if found, None otherwise.

update_permission_rank `async` ¶

Python

update_permission_rank(
    guild_id: int, rank: int, name: str | None = None, description: str | None = None
) -> PermissionRank | None

Update a permission rank.

Returns:

PermissionRank | None –

The updated permission rank, or None if not found.

delete_permission_rank `async` ¶

Python

delete_permission_rank(guild_id: int, rank: int) -> bool

Delete a permission rank.

Returns:

bool –

True if deleted successfully, False otherwise.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

ReminderController ¶

Python

ReminderController(db: DatabaseService | None = None)

Bases: BaseController[Reminder]

Clean Reminder controller using the new BaseController pattern.

Initialize the reminder controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes reminder

Methods:

get_reminder_by_id –

Get a reminder by its ID.
get_reminders_by_user –

Get all reminders for a specific user in a guild.
get_reminders_by_guild –

Get all reminders in a guild.
create_reminder –

Create a new reminder.
update_reminder –

Update a reminder by ID.
delete_reminder –

Delete a reminder by ID.
get_expired_reminders –

Get all expired reminders.
get_active_reminders –

Get all active (non-expired) reminders in a guild.
get_reminders_by_channel –

Get all reminders for a specific channel.
get_reminder_count_by_user –

Get the number of reminders for a user in a guild.
get_reminder_count_by_guild –

Get the total number of reminders in a guild.
delete_reminder_by_id –

Delete a reminder by its ID.
get_all_reminders –

Get all reminders in a guild.
insert_reminder –

Insert a new reminder - alias for create.
cleanup_expired_reminders –

Delete all expired reminders and return the count.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_reminder_by_id `async` ¶

Python

get_reminder_by_id(reminder_id: int) -> Reminder | None

Get a reminder by its ID.

Returns:

Reminder | None –

The reminder if found, None otherwise.

get_reminders_by_user `async` ¶

Python

get_reminders_by_user(user_id: int, guild_id: int) -> list[Reminder]

Get all reminders for a specific user in a guild.

Returns:

list[Reminder] –

List of all reminders for the user in the guild.

get_reminders_by_guild `async` ¶

Python

get_reminders_by_guild(guild_id: int) -> list[Reminder]

Get all reminders in a guild.

Returns:

list[Reminder] –

List of all reminders for the guild.

create_reminder `async` ¶

Python

create_reminder(
    user_id: int,
    guild_id: int,
    channel_id: int,
    message: str,
    expires_at: datetime,
    **kwargs: Any,
) -> Reminder

Create a new reminder.

Returns:

Reminder –

The newly created reminder.

update_reminder `async` ¶

Python

update_reminder(reminder_id: int, **kwargs: Any) -> Reminder | None

Update a reminder by ID.

Returns:

Reminder | None –

The updated reminder, or None if not found.

delete_reminder `async` ¶

Python

delete_reminder(reminder_id: int) -> bool

Delete a reminder by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

get_expired_reminders `async` ¶

Python

get_expired_reminders() -> list[Reminder]

Get all expired reminders.

Returns:

list[Reminder] –

List of all expired reminders.

get_active_reminders `async` ¶

Python

get_active_reminders(guild_id: int) -> list[Reminder]

Get all active (non-expired) reminders in a guild.

Returns:

list[Reminder] –

List of active reminders.

get_reminders_by_channel `async` ¶

Python

get_reminders_by_channel(channel_id: int) -> list[Reminder]

Get all reminders for a specific channel.

Returns:

list[Reminder] –

List of reminders for the channel.

get_reminder_count_by_user `async` ¶

Python

get_reminder_count_by_user(user_id: int, guild_id: int) -> int

Get the number of reminders for a user in a guild.

Returns:

int –

The count of reminders for the user.

get_reminder_count_by_guild `async` ¶

Python

get_reminder_count_by_guild(guild_id: int) -> int

Get the total number of reminders in a guild.

Returns:

int –

The total count of reminders in the guild.

delete_reminder_by_id `async` ¶

Python

delete_reminder_by_id(reminder_id: int) -> bool

Delete a reminder by its ID.

Returns:

bool –

True if deleted successfully, False otherwise.

get_all_reminders `async` ¶

Python

get_all_reminders(guild_id: int) -> list[Reminder]

Get all reminders in a guild.

Returns:

list[Reminder] –

List of all reminders for the guild.

insert_reminder `async` ¶

Python

insert_reminder(**kwargs: Any) -> Reminder

Insert a new reminder - alias for create.

Returns:

Reminder –

The newly created reminder.

cleanup_expired_reminders `async` ¶

Python

cleanup_expired_reminders() -> int

Delete all expired reminders and return the count.

Returns:

int –

The number of reminders that were deleted.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

SnippetController ¶

Python

SnippetController(db: DatabaseService | None = None)

Bases: BaseController[Snippet]

Clean Snippet controller using the new BaseController pattern.

Initialize the snippet controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes snippet

Methods:

get_snippet_by_id –

Get a snippet by its ID.
get_snippet_by_name_and_guild –

Get a snippet by name and guild.
get_snippets_by_guild –

Get all snippets in a guild.
create_snippet –

Create a new snippet.
update_snippet –

Update a snippet by ID.
update_snippet_by_id –

Update a snippet by ID - alias for update_snippet.
delete_snippet –

Delete a snippet by ID.
delete_snippet_by_id –

Delete a snippet by ID - alias for delete_snippet.
get_snippets_by_creator –

Get all snippets created by a specific user in a guild.
search_snippets –

Search snippets by name or content in a guild.
get_snippet_count_by_guild –

Get the total number of snippets in a guild.
find_many –

Find many snippets with optional filters - alias for find_all.
get_snippet_by_name_and_guild_id –

Get a snippet by name and guild ID.
create_snippet_alias –

Create a snippet alias.
get_snippet_count_by_creator –

Get the number of snippets created by a user in a guild.
toggle_snippet_lock –

Toggle the locked status of a snippet.
toggle_snippet_lock_by_id –

Toggle the locked status of a snippet by ID - alias for toggle_snippet_lock.
increment_snippet_uses –

Increment the usage count of a snippet.
get_popular_snippets –

Get the most popular snippets in a guild by usage count.
get_snippets_by_alias –

Get snippets by alias in a guild.
get_all_aliases –

Get all aliases in a guild.
get_all_snippets_by_guild_id –

Get all snippets in a guild - alias for get_snippets_by_guild.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_snippet_by_id `async` ¶

Python

get_snippet_by_id(snippet_id: int) -> Snippet | None

Get a snippet by its ID.

Returns:

Snippet | None –

The snippet if found, None otherwise.

get_snippet_by_name_and_guild `async` ¶

Python

get_snippet_by_name_and_guild(snippet_name: str, guild_id: int) -> Snippet | None

Get a snippet by name and guild.

Returns:

Snippet | None –

The snippet if found, None otherwise.

get_snippets_by_guild `async` ¶

Python

get_snippets_by_guild(guild_id: int) -> list[Snippet]

Get all snippets in a guild.

Returns:

list[Snippet] –

List of all snippets for the guild.

create_snippet `async` ¶

Python

create_snippet(
    snippet_name: str,
    snippet_content: str,
    guild_id: int,
    snippet_user_id: int,
    alias: str | None = None,
    **kwargs: Any,
) -> Snippet

Create a new snippet.

Returns:

Snippet –

The newly created snippet.

update_snippet `async` ¶

Python

update_snippet(snippet_id: int, **kwargs: Any) -> Snippet | None

Update a snippet by ID.

Returns:

Snippet | None –

The updated snippet, or None if not found.

update_snippet_by_id `async` ¶

Python

update_snippet_by_id(snippet_id: int, **kwargs: Any) -> Snippet | None

Update a snippet by ID - alias for update_snippet.

Returns:

Snippet | None –

The updated snippet, or None if not found.

delete_snippet `async` ¶

Python

delete_snippet(snippet_id: int) -> bool

Delete a snippet by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

delete_snippet_by_id `async` ¶

Python

delete_snippet_by_id(snippet_id: int) -> bool

Delete a snippet by ID - alias for delete_snippet.

Returns:

bool –

True if deleted successfully, False otherwise.

get_snippets_by_creator `async` ¶

Python

get_snippets_by_creator(creator_id: int, guild_id: int) -> list[Snippet]

Get all snippets created by a specific user in a guild.

Returns:

list[Snippet] –

List of snippets created by the user.

search_snippets `async` ¶

Python

search_snippets(guild_id: int, search_term: str) -> list[Snippet]

Search snippets by name or content in a guild.

Returns:

list[Snippet] –

List of snippets matching the search term.

get_snippet_count_by_guild `async` ¶

Python

get_snippet_count_by_guild(guild_id: int) -> int

Get the total number of snippets in a guild.

Returns:

int –

The total count of snippets in the guild.

find_many `async` ¶

Python

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

Find many snippets with optional filters - alias for find_all.

Returns:

list[Snippet] –

List of snippets matching the filters.

get_snippet_by_name_and_guild_id `async` ¶

Python

get_snippet_by_name_and_guild_id(name: str, guild_id: int) -> Snippet | None

Get a snippet by name and guild ID.

Returns:

Snippet | None –

The snippet if found, None otherwise.

create_snippet_alias `async` ¶

Python

create_snippet_alias(original_name: str, alias_name: str, guild_id: int) -> Snippet

Create a snippet alias.

Returns:

Snippet –

The newly created alias snippet.

Raises:

ValueError –

If the original snippet is not found.

get_snippet_count_by_creator `async` ¶

Python

get_snippet_count_by_creator(creator_id: int, guild_id: int) -> int

Get the number of snippets created by a user in a guild.

Returns:

int –

The count of snippets created by the user.

toggle_snippet_lock `async` ¶

Python

toggle_snippet_lock(snippet_id: int) -> Snippet | None

Toggle the locked status of a snippet.

Returns:

Snippet | None –

The updated snippet, or None if not found.

toggle_snippet_lock_by_id `async` ¶

Python

toggle_snippet_lock_by_id(snippet_id: int) -> Snippet | None

Toggle the locked status of a snippet by ID - alias for toggle_snippet_lock.

Returns:

Snippet | None –

The updated snippet, or None if not found.

increment_snippet_uses `async` ¶

Python

increment_snippet_uses(snippet_id: int) -> Snippet | None

Increment the usage count of a snippet.

Returns:

Snippet | None –

The updated snippet, or None if not found.

get_popular_snippets `async` ¶

Python

get_popular_snippets(guild_id: int, limit: int = 10) -> list[Snippet]

Get the most popular snippets in a guild by usage count.

Returns:

list[Snippet] –

List of snippets sorted by usage count (most popular first).

get_snippets_by_alias `async` ¶

Python

get_snippets_by_alias(alias: str, guild_id: int) -> list[Snippet]

Get snippets by alias in a guild.

Returns:

list[Snippet] –

List of snippets with the specified alias.

get_all_aliases `async` ¶

Python

get_all_aliases(guild_id: int) -> list[Snippet]

Get all aliases in a guild.

Returns:

list[Snippet] –

List of all alias snippets.

get_all_snippets_by_guild_id `async` ¶

Python

get_all_snippets_by_guild_id(guild_id: int) -> list[Snippet]

Get all snippets in a guild - alias for get_snippets_by_guild.

Returns:

list[Snippet] –

List of all snippets for the guild.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

StarboardController ¶

Python

StarboardController(db: DatabaseService | None = None)

Bases: BaseController[Starboard]

Clean Starboard controller using the new BaseController pattern.

Initialize the starboard controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes starboard

Methods:

get_starboard_by_guild –

Get starboard configuration for a guild.
get_or_create_starboard –

Get starboard configuration, or create it with defaults if it doesn't exist.
update_starboard –

Update starboard configuration.
delete_starboard –

Delete starboard configuration for a guild.
get_all_starboards –

Get all starboard configurations.
get_starboard_count –

Get the total number of starboard configurations.
create_or_update_starboard –

Create or update starboard configuration for a guild.
delete_starboard_by_guild_id –

Delete starboard configuration for a guild.
get_starboard_by_guild_id –

Get starboard configuration by guild ID - alias for get_starboard_by_guild.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_starboard_by_guild `async` ¶

Python

get_starboard_by_guild(guild_id: int) -> Starboard | None

Get starboard configuration for a guild.

Returns:

Starboard | None –

The starboard configuration if found, None otherwise.

get_or_create_starboard `async` ¶

Python

get_or_create_starboard(guild_id: int, **defaults: Any) -> Starboard

Get starboard configuration, or create it with defaults if it doesn't exist.

Returns:

Starboard –

The starboard configuration (existing or newly created).

update_starboard `async` ¶

Python

update_starboard(guild_id: int, **updates: Any) -> Starboard | None

Update starboard configuration.

Returns:

Starboard | None –

The updated starboard configuration, or None if not found.

delete_starboard `async` ¶

Python

delete_starboard(guild_id: int) -> bool

Delete starboard configuration for a guild.

Returns:

bool –

True if deleted successfully, False otherwise.

get_all_starboards `async` ¶

Python

get_all_starboards() -> list[Starboard]

Get all starboard configurations.

Returns:

list[Starboard] –

List of all starboard configurations.

get_starboard_count `async` ¶

Python

get_starboard_count() -> int

Get the total number of starboard configurations.

Returns:

int –

The total count of starboard configurations.

create_or_update_starboard `async` ¶

Python

create_or_update_starboard(guild_id: int, **kwargs: Any) -> Starboard

Create or update starboard configuration for a guild.

Returns:

Starboard –

The starboard configuration (created or updated).

delete_starboard_by_guild_id `async` ¶

Python

delete_starboard_by_guild_id(guild_id: int) -> bool

Delete starboard configuration for a guild.

Returns:

bool –

True if deleted successfully, False otherwise.

get_starboard_by_guild_id `async` ¶

Python

get_starboard_by_guild_id(guild_id: int) -> Starboard | None

Get starboard configuration by guild ID - alias for get_starboard_by_guild.

Returns:

Starboard | None –

The starboard configuration if found, None otherwise.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

StarboardMessageController ¶

Python

StarboardMessageController(db: DatabaseService | None = None)

Bases: BaseController[StarboardMessage]

Clean StarboardMessage controller using the new BaseController pattern.

Initialize the starboard message controller.

Parameters:

db (DatabaseService | None, default: None ) –

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

Returned by:

controllers Classes DatabaseCoordinator Attributes starboard_message

Methods:

get_message_by_id –

Get a starboard message by its ID.
get_message_by_original –

Get a starboard message by its original message ID and guild.
get_messages_by_guild –

Get all starboard messages in a guild.
create_starboard_message –

Create a new starboard message.
update_star_count –

Update the star count for a starboard message.
delete_starboard_message –

Delete a starboard message.
get_top_messages –

Get top starboard messages by star count in a guild.
get_message_count_by_guild –

Get the total number of starboard messages in a guild.
get_messages_by_channel –

Get all starboard messages in a specific channel.
get_starboard_message_by_id –

Get a starboard message by its ID.
create_or_update_starboard_message –

Create or update a starboard message.
create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
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_message_by_id `async` ¶

Python

get_message_by_id(message_id: int) -> StarboardMessage | None

Get a starboard message by its ID.

Returns:

StarboardMessage | None –

The starboard message if found, None otherwise.

get_message_by_original `async` ¶

Python

get_message_by_original(
    original_message_id: int, guild_id: int
) -> StarboardMessage | None

Get a starboard message by its original message ID and guild.

Returns:

StarboardMessage | None –

The starboard message if found, None otherwise.

get_messages_by_guild `async` ¶

Python

get_messages_by_guild(
    guild_id: int, limit: int | None = None
) -> list[StarboardMessage]

Get all starboard messages in a guild.

Returns:

list[StarboardMessage] –

List of starboard messages sorted by star count (limited if specified).

create_starboard_message `async` ¶

Python

create_starboard_message(
    original_message_id: int,
    starboard_message_id: int,
    guild_id: int,
    channel_id: int,
    star_count: int = 1,
    **kwargs: Any,
) -> StarboardMessage

Create a new starboard message.

Returns:

StarboardMessage –

The newly created starboard message.

update_star_count `async` ¶

Python

update_star_count(message_id: int, new_star_count: int) -> StarboardMessage | None

Update the star count for a starboard message.

Returns:

StarboardMessage | None –

The updated starboard message, or None if not found.

delete_starboard_message `async` ¶

Python

delete_starboard_message(message_id: int) -> bool

Delete a starboard message.

Returns:

bool –

True if deleted successfully, False otherwise.

get_top_messages `async` ¶

Python

get_top_messages(guild_id: int, limit: int = 10) -> list[StarboardMessage]

Get top starboard messages by star count in a guild.

Returns:

list[StarboardMessage] –

List of top starboard messages sorted by star count.

get_message_count_by_guild `async` ¶

Python

get_message_count_by_guild(guild_id: int) -> int

Get the total number of starboard messages in a guild.

Returns:

int –

The total count of starboard messages in the guild.

get_messages_by_channel `async` ¶

Python

get_messages_by_channel(channel_id: int) -> list[StarboardMessage]

Get all starboard messages in a specific channel.

Returns:

list[StarboardMessage] –

List of all starboard messages in the channel.

get_starboard_message_by_id `async` ¶

Python

get_starboard_message_by_id(message_id: int) -> StarboardMessage | None

Get a starboard message by its ID.

Returns:

StarboardMessage | None –

The starboard message if found, None otherwise.

create_or_update_starboard_message `async` ¶

Python

create_or_update_starboard_message(**kwargs: Any) -> StarboardMessage

Create or update a starboard message.

Returns:

StarboardMessage –

The starboard message (created or updated).

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

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.

DatabaseCoordinator ¶

Python

DatabaseCoordinator(db: DatabaseService | None = None)

Coordinator for database controllers with lazy loading.

Provides centralized access to all database controllers with lazy initialization to avoid unnecessary resource allocation. Acts as a facade for database operations.

Parameters:

db (DatabaseService, default: None ) –

The database service instance to use for operations.

Attributes:

db (DatabaseService) –

The underlying database service.

Raises:

RuntimeError –

If no database service is provided.

Initialize the database coordinator.

Parameters:

db (DatabaseService, default: None ) –

The database service instance. If None, raises RuntimeError.

Raises:

RuntimeError –

If no database service is provided.

Returned by:

Used by:

Attributes¶

guild `property` ¶

Python

guild: GuildController

Get the guild controller for guild-related operations.

guild_config `property` ¶

Python

guild_config: GuildConfigController

Get the guild configuration controller.

afk `property` ¶

Python

afk: AfkController

Get the AFK status controller.

levels `property` ¶

Python

levels: LevelsController

Get the user leveling controller.

snippet `property` ¶

Python

snippet: SnippetController

Get the snippet controller for custom commands.

case `property` ¶

Python

case: CaseController

Get the moderation case controller.

starboard `property` ¶

Python

starboard: StarboardController

Get the starboard configuration controller.

starboard_message `property` ¶

Python

starboard_message: StarboardMessageController

Get the starboard message controller.

reminder `property` ¶

Python

reminder: ReminderController

Get the reminder controller.

permission_ranks `property` ¶

Python

permission_ranks: PermissionRankController

Get the permission ranks controller.

permission_assignments `property` ¶

Python

permission_assignments: PermissionAssignmentController

Get the permission assignments controller.

command_permissions `property` ¶

Python

command_permissions: PermissionCommandController

Get the command permissions controller.

controllers ¶

Classes¶

AfkController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

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 ¶

set_afk async ¶

find_many async ¶

is_perm_afk async ¶

get_expired_afk_members async ¶

_get_pagination ¶

_get_bulk ¶

_get_transaction ¶

_get_performance ¶

_get_upsert ¶

create async ¶

get_by_id async ¶

update_by_id async ¶

delete_by_id async ¶

exists 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 ¶

BaseController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

_get_pagination ¶

_get_bulk ¶

_get_transaction ¶

_get_performance ¶

_get_upsert ¶

create async ¶

get_by_id async ¶

update_by_id async ¶

delete_by_id async ¶

exists 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 ¶

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` ¶

set_afk `async` ¶

find_many `async` ¶

is_perm_afk `async` ¶

get_expired_afk_members `async` ¶

create `async` ¶

get_by_id `async` ¶

update_by_id `async` ¶

delete_by_id `async` ¶

exists `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` ¶

create `async` ¶

get_by_id `async` ¶

update_by_id `async` ¶

delete_by_id `async` ¶

exists `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` ¶