levels ¶

User level and XP tracking controller.

This controller manages Discord user experience points, levels, and ranking information for guilds, supporting features like leveling systems and leaderboards.

Classes:

LevelsController –

Clean Levels controller using the new BaseController pattern.

Classes¶

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

Create a new record.
get_level –

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

Get a record by ID.
update_xp_and_level –

Update XP and level for a member.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
reset_xp –

Reset XP and level for a member.
find_one –

Find one record.
find_all –

Find all records with performance optimizations.
toggle_blacklist –

Toggle blacklist status for a member.
find_all_with_options –

Find all records with relationship loading options.
is_blacklisted –

Check if a member is blacklisted.
count –

Count records.
get_last_message_time –

Get the last message time for a member.
get_all –

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

Get both XP and level for a member.
execute_query –

Execute a custom query.
get_member_rank –

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

Find records using JSON column queries.
get_guild_stats –

Get guild statistics.
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.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

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.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

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.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

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

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

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

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

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.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

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.

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

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.

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.

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.

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.

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.

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.

count `async` ¶

Python

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

Count records.

Returns:

int –

The count of matching records.

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

execute_query `async` ¶

Python

execute_query(query: Any) -> Any

Execute a custom query.

Returns:

Any –

The query result.

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.

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.

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.

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.

levels ¶

Classes¶

LevelsController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

get_levels_by_member async ¶

get_or_create_levels async ¶

add_xp async ¶

set_xp async ¶

_get_pagination ¶

set_level async ¶

_get_bulk ¶

blacklist_member async ¶

_get_transaction ¶

unblacklist_member async ¶

_get_performance ¶

get_top_members async ¶

_get_upsert ¶

get_xp async ¶

create async ¶

get_level async ¶

get_by_id async ¶

update_xp_and_level async ¶

update_by_id async ¶

delete_by_id async ¶

exists async ¶

reset_xp async ¶

find_one async ¶

find_all async ¶

toggle_blacklist async ¶

find_all_with_options async ¶

is_blacklisted async ¶

count async ¶

get_last_message_time async ¶

get_all async ¶

get_xp_and_level async ¶

execute_query async ¶

get_member_rank async ¶

find_with_json_query async ¶

get_guild_stats async ¶

find_with_array_contains async ¶

find_with_full_text_search async ¶

paginate async ¶

find_paginated async ¶

bulk_create async ¶

bulk_update async ¶

bulk_delete async ¶

update_where async ¶

delete_where async ¶

bulk_upsert_with_conflict_resolution async ¶

with_session async ¶

with_transaction async ¶

execute_transaction async ¶

get_table_statistics async ¶

explain_query_performance async ¶

upsert_by_field async ¶

upsert_by_id async ¶

get_or_create_by_field async ¶

get_or_create async ¶

upsert async ¶

db_service `property` ¶

model_class `property` ¶

get_levels_by_member `async` ¶

get_or_create_levels `async` ¶

add_xp `async` ¶

set_xp `async` ¶

set_level `async` ¶

blacklist_member `async` ¶

unblacklist_member `async` ¶

get_top_members `async` ¶

get_xp `async` ¶

create `async` ¶

get_level `async` ¶

get_by_id `async` ¶

update_xp_and_level `async` ¶

update_by_id `async` ¶

delete_by_id `async` ¶

exists `async` ¶

reset_xp `async` ¶

find_one `async` ¶

find_all `async` ¶

toggle_blacklist `async` ¶

find_all_with_options `async` ¶

is_blacklisted `async` ¶

count `async` ¶

get_last_message_time `async` ¶

get_all `async` ¶

get_xp_and_level `async` ¶

execute_query `async` ¶

get_member_rank `async` ¶

find_with_json_query `async` ¶

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