permissions ¶

Dynamic permission system controllers.

Provides database operations for the flexible permission system that allows servers to customize their permission levels and role assignments.

Classes:

PermissionRankController –

Controller for managing guild permission ranks.
PermissionAssignmentController –

Controller for managing guild permission assignments.
PermissionCommandController –

Controller for managing command permission requirements.

Classes¶

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.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

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.

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

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

Create a new record.
get_by_id –

Get a record by ID.
get_assignments_by_guild –

Get all permission assignments for a guild.
update_by_id –

Update a record by ID.
remove_role_assignment –

Remove a permission level assignment from a role.
delete_by_id –

Delete a record by ID.
get_user_permission_rank –

Get the highest permission rank a user has based on their roles.
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.

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.

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.

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.

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.

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.

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.

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

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:

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

Set the permission rank required for a command.
count –

Count records.
get_all –

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

Get the permission requirement for a specific command.
execute_query –

Execute a custom query.
get_all_command_permissions –

Get all command permissions for a guild.
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.

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

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.

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.

execute_query `async` ¶

Python

execute_query(query: Any) -> Any

Execute a custom query.

Returns:

Any –

The query result.

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.

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.

permissions ¶

Classes¶

PermissionRankController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

create_permission_rank async ¶

get_permission_ranks_by_guild async ¶

get_permission_rank async ¶

_get_pagination ¶

update_permission_rank async ¶

_get_bulk ¶

_get_transaction ¶

delete_permission_rank async ¶

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

PermissionAssignmentController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

_get_pagination ¶

_get_bulk ¶

_get_transaction ¶

_get_performance ¶

_get_upsert ¶

assign_permission_rank async ¶

create async ¶

get_by_id async ¶

get_assignments_by_guild async ¶

update_by_id async ¶

remove_role_assignment async ¶

delete_by_id async ¶

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

db_service `property` ¶

model_class `property` ¶

create_permission_rank `async` ¶

get_permission_ranks_by_guild `async` ¶

get_permission_rank `async` ¶

update_permission_rank `async` ¶

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

assign_permission_rank `async` ¶

create `async` ¶

get_by_id `async` ¶

get_assignments_by_guild `async` ¶

update_by_id `async` ¶

remove_role_assignment `async` ¶

delete_by_id `async` ¶

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