tux.database.controllers.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:

Name	Description
LevelsController	Clean Levels controller using the new BaseController pattern.

Classes

LevelsController(db: DatabaseService | None = None)

Bases: BaseController[Levels]

Clean Levels controller using the new BaseController pattern.

Initialize the levels controller.

Parameters:

Name	Type	Description	Default
db	DatabaseService | None	The database service instance. If None, uses the default service.	None

Methods:

Name	Description
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:

Name	Type	Description
db_service	DatabaseService	Database service property for test compatibility.
model_class	type[ModelT]	Model class property for test compatibility.

Source code in tux/database/controllers/levels.py

Python

def __init__(self, db: DatabaseService | None = None) -> None:
    """Initialize the levels controller.

    Parameters
    ----------
    db : DatabaseService | None, optional
        The database service instance. If None, uses the default service.
    """
    super().__init__(Levels, db)

Attributes

db_service: DatabaseService property

Database service property for test compatibility.

Returns:

Type	Description
DatabaseService	The database service instance.

model_class: type[ModelT] property

Model class property for test compatibility.

Returns:

Type	Description
type[ModelT]	The SQLModel class.

Functions

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

Get levels for a specific member in a guild.

Returns:

Type	Description
Levels | None	The levels record if found, None otherwise.

Source code in tux/database/controllers/levels.py

Python

async def get_levels_by_member(self, 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.
    """
    return await self.find_one(filters=(Levels.member_id == member_id) & (Levels.guild_id == guild_id))

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

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

Returns:

Type	Description
Levels	The levels record for the member.

Source code in tux/database/controllers/levels.py

Python

async def get_or_create_levels(self, 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.
    """
    levels = await self.get_levels_by_member(member_id, guild_id)
    if levels is not None:
        return levels
    return await self.create(
        member_id=member_id,
        guild_id=guild_id,
        xp=0.0,
        level=0,
        blacklisted=False,
        last_message=datetime.now(UTC),
    )

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

Add XP to a member's levels.

Returns:

Type	Description
Levels	The updated levels record.

Source code in tux/database/controllers/levels.py

Python

async def add_xp(self, member_id: int, guild_id: int, xp_amount: float) -> Levels:
    """
    Add XP to a member's levels.

    Returns
    -------
    Levels
        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    new_xp = levels.xp + xp_amount
    new_level = int(new_xp**0.5)  # Simple level calculation

    return (
        await self.update_by_id(levels.member_id, xp=new_xp, level=new_level, last_message=datetime.now(UTC))
        or levels
    )

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

Set a member's XP to a specific value.

Returns:

Type	Description
Levels	The updated levels record.

Source code in tux/database/controllers/levels.py

Python

async def set_xp(self, member_id: int, guild_id: int, xp: float) -> Levels:
    """
    Set a member's XP to a specific value.

    Returns
    -------
    Levels
        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    new_level = int(xp**0.5)

    return (
        await self.update_by_id(levels.member_id, xp=xp, level=new_level, last_message=datetime.now(UTC)) or levels
    )

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

Type	Description
PaginationController[ModelT]	The pagination controller instance.

Source code in tux/database/controllers/levels.py

Python

    new_level = int(xp**0.5)

    return (
        await self.update_by_id(levels.member_id, xp=xp, level=new_level, last_message=datetime.now(UTC)) or levels
    )

async def set_level(self, member_id: int, guild_id: int, level: int) -> Levels:
    """
    Set a member's level to a specific value.

    Returns
    -------

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

Set a member's level to a specific value.

Returns:

Type	Description
Levels	The updated levels record.

Source code in tux/database/controllers/levels.py

Python

async def set_level(self, member_id: int, guild_id: int, level: int) -> Levels:
    """
    Set a member's level to a specific value.

    Returns
    -------
    Levels
        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    xp = level**2  # Reverse level calculation

    return await self.update_by_id(levels.member_id, xp=xp, level=level, last_message=datetime.now(UTC)) or levels

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

Type	Description
BulkOperationsController[ModelT]	The bulk operations controller instance.

Source code in tux/database/controllers/levels.py

Python

        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    xp = level**2  # Reverse level calculation

    return await self.update_by_id(levels.member_id, xp=xp, level=level, last_message=datetime.now(UTC)) or levels

async def blacklist_member(self, member_id: int, guild_id: int) -> Levels:
    """
    Blacklist a member from gaining XP.

    Returns

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

Blacklist a member from gaining XP.

Returns:

Type	Description
Levels	The updated levels record.

Source code in tux/database/controllers/levels.py

Python

async def blacklist_member(self, member_id: int, guild_id: int) -> Levels:
    """
    Blacklist a member from gaining XP.

    Returns
    -------
    Levels
        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return await self.update_by_id(levels.member_id, blacklisted=True) or levels

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

Type	Description
TransactionController[ModelT]	The transaction controller instance.

Source code in tux/database/controllers/levels.py

Python

    Levels
        The updated levels record.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return await self.update_by_id(levels.member_id, blacklisted=True) or levels

async def unblacklist_member(self, member_id: int, guild_id: int) -> Levels:
    """
    Remove a member from the blacklist.

    Returns
    -------

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

Remove a member from the blacklist.

Returns:

Type	Description
Levels	The updated levels record.

Source code in tux/database/controllers/levels.py

Python

async def unblacklist_member(self, member_id: int, guild_id: int) -> Levels:
    """
    Remove a member from the blacklist.

    Returns
    -------
    Levels
        The updated levels record.
    """
    levels = await self.get_levels_by_member(member_id, guild_id)
    if levels is None:
        return await self.get_or_create_levels(member_id, guild_id)
    return await self.update_by_id(levels.member_id, blacklisted=False) or levels

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

Type	Description
PerformanceController[ModelT]	The performance controller instance.

Source code in tux/database/controllers/levels.py

Python

        The updated levels record.
    """
    levels = await self.get_levels_by_member(member_id, guild_id)
    if levels is None:
        return await self.get_or_create_levels(member_id, guild_id)
    return await self.update_by_id(levels.member_id, blacklisted=False) or levels

async def get_top_members(self, guild_id: int, limit: int = 10) -> list[Levels]:
    """
    Get top members by XP in a guild.

    Returns

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

Get top members by XP in a guild.

Returns:

Type	Description
list[Levels]	List of top members sorted by XP (highest first).

Source code in tux/database/controllers/levels.py

Python

async def get_top_members(self, 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).
    """
    all_members = await self.find_all(filters=Levels.guild_id == guild_id)
    # Sort by XP descending and limit
    sorted_members = sorted(all_members, key=lambda x: x.xp, reverse=True)
    return sorted_members[:limit]

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

Type	Description
UpsertController[ModelT]	The upsert controller instance.

Source code in tux/database/controllers/levels.py

Python

    list[Levels]
        List of top members sorted by XP (highest first).
    """
    all_members = await self.find_all(filters=Levels.guild_id == guild_id)
    # Sort by XP descending and limit
    sorted_members = sorted(all_members, key=lambda x: x.xp, reverse=True)
    return sorted_members[:limit]

# Additional methods that module files expect
async def get_xp(self, member_id: int, guild_id: int) -> float:
    """
    Get XP for a specific member in a guild.

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

Get XP for a specific member in a guild.

Returns:

Type	Description
float	The member's XP value.

Source code in tux/database/controllers/levels.py

Python

async def get_xp(self, member_id: int, guild_id: int) -> float:
    """
    Get XP for a specific member in a guild.

    Returns
    -------
    float
        The member's XP value.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.xp

create(**kwargs: Any) -> ModelT async

Create a new record.

Returns:

Type	Description
ModelT	The newly created record.

Source code in tux/database/controllers/levels.py

Python

    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.xp

async def get_level(self, member_id: int, guild_id: int) -> int:
    """
    Get level for a specific member in a guild.

    Returns
    -------

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

Get level for a specific member in a guild.

Returns:

Type	Description
int	The member's level.

Source code in tux/database/controllers/levels.py

Python

async def get_level(self, member_id: int, guild_id: int) -> int:
    """
    Get level for a specific member in a guild.

    Returns
    -------
    int
        The member's level.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.level

get_by_id(record_id: Any) -> ModelT | None async

Get a record by ID.

Returns:

Type	Description
ModelT | None	The record if found, None otherwise.

Source code in tux/database/controllers/levels.py

Python

        The member's level.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.level

async def update_xp_and_level(
    self,
    member_id: int,
    guild_id: int,
    xp_amount: float | None = None,

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 async

Update XP and level for a member.

Returns:

Type	Description
Levels	The updated levels record.

Raises:

Type	Description
ValueError	If xp_amount, new_level, or last_message is missing.

Source code in tux/database/controllers/levels.py

Python

async def update_xp_and_level(
    self,
    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.
    """
    # Handle both positional and named parameter styles
    if xp_amount is None and "xp" in kwargs:
        xp_amount = kwargs["xp"]
    if new_level is None and "level" in kwargs:
        new_level = kwargs["level"]
    if last_message is None and "last_message" in kwargs:
        last_message = kwargs["last_message"]

    if xp_amount is None or new_level is None or last_message is None:
        error_msg = "xp_amount, new_level, and last_message are required"
        raise ValueError(error_msg)

    # Use composite key for update
    await self.update_where(
        (Levels.member_id == member_id) & (Levels.guild_id == guild_id),
        {"xp": xp_amount, "level": new_level, "last_message": last_message},
    )
    # Return updated record
    return await self.get_or_create_levels(member_id, guild_id)

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

Update a record by ID.

Returns:

Type	Description
ModelT | None	The updated record, or None if not found.

Source code in tux/database/controllers/levels.py

Python

    last_message: datetime | None = None,
    **kwargs: Any,
) -> Levels:
    """
    Update XP and level for a member.

    Returns
    -------
    Levels
        The updated levels record.

delete_by_id(record_id: Any) -> bool async

Delete a record by ID.

Returns:

Type	Description
bool	True if deleted successfully, False otherwise.

Source code in tux/database/controllers/levels.py

Python

Raises
------
ValueError
    If xp_amount, new_level, or last_message is missing.
"""
# Handle both positional and named parameter styles
if xp_amount is None and "xp" in kwargs:
    xp_amount = kwargs["xp"]
if new_level is None and "level" in kwargs:
    new_level = kwargs["level"]

exists(filters: Any) -> bool async

Check if a record exists.

Returns:

Type	Description
bool	True if record exists, False otherwise.

Source code in tux/database/controllers/levels.py

Python

    last_message = kwargs["last_message"]

if xp_amount is None or new_level is None or last_message is None:
    error_msg = "xp_amount, new_level, and last_message are required"
    raise ValueError(error_msg)

# Use composite key for update
await self.update_where(
    (Levels.member_id == member_id) & (Levels.guild_id == guild_id),
    {"xp": xp_amount, "level": new_level, "last_message": last_message},

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

Reset XP and level for a member.

Returns:

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

Source code in tux/database/controllers/levels.py

Python

async def reset_xp(self, 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.
    """
    # Use composite key for update
    await self.update_where(
        (Levels.member_id == member_id) & (Levels.guild_id == guild_id),
        {"xp": 0.0, "level": 0},
    )
    # Return updated record
    return await self.get_or_create_levels(member_id, guild_id)

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

Find one record.

Returns:

Type	Description
ModelT | None	The found record, or None if not found.

Source code in tux/database/controllers/levels.py

Python

"""
Reset XP and level for a member.

Returns
-------
Levels
    The updated levels record with XP and level reset to 0.
"""
# Use composite key for update
await self.update_where(

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

Find all records with performance optimizations.

Returns:

Type	Description
list[ModelT]	List of found records.

Source code in tux/database/controllers/levels.py

Python

        {"xp": 0.0, "level": 0},
    )
    # Return updated record
    return await self.get_or_create_levels(member_id, guild_id)

async def toggle_blacklist(self, member_id: int, guild_id: int) -> bool:
    """
    Toggle blacklist status for a member.

    Returns
    -------
    bool
        The new blacklist status.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    new_status = not levels.blacklisted

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

Toggle blacklist status for a member.

Returns:

Type	Description
bool	The new blacklist status.

Source code in tux/database/controllers/levels.py

Python

async def toggle_blacklist(self, member_id: int, guild_id: int) -> bool:
    """
    Toggle blacklist status for a member.

    Returns
    -------
    bool
        The new blacklist status.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    new_status = not levels.blacklisted
    # Use composite key for update
    await self.update_where(
        (Levels.member_id == member_id) & (Levels.guild_id == guild_id),
        {"blacklisted": new_status},
    )
    return new_status

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

Find all records with relationship loading options.

Returns:

Type	Description
list[ModelT]	List of found records with loaded relationships.

Source code in tux/database/controllers/levels.py

Python

    await self.update_where(
        (Levels.member_id == member_id) & (Levels.guild_id == guild_id),
        {"blacklisted": new_status},
    )
    return new_status

# Additional methods that module files expect
async def is_blacklisted(self, member_id: int, guild_id: int) -> bool:
    """
    Check if a member is blacklisted.

    Returns
    -------
    bool
        True if member is blacklisted, False otherwise.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)

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

Check if a member is blacklisted.

Returns:

Type	Description
bool	True if member is blacklisted, False otherwise.

Source code in tux/database/controllers/levels.py

Python

async def is_blacklisted(self, member_id: int, guild_id: int) -> bool:
    """
    Check if a member is blacklisted.

    Returns
    -------
    bool
        True if member is blacklisted, False otherwise.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.blacklisted

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

Count records.

Returns:

Type	Description
int	The count of matching records.

Source code in tux/database/controllers/levels.py

Python

async def get_last_message_time(self, 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_last_message_time(member_id: int, guild_id: int) -> datetime async

Get the last message time for a member.

Returns:

Type	Description
datetime	The timestamp of the member's last message.

Source code in tux/database/controllers/levels.py

Python

async def get_last_message_time(self, 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.
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.last_message

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

Get all records (alias for find_all without pagination).

Returns:

Type	Description
list[ModelT]	List of all matching records.

Source code in tux/database/controllers/levels.py

Python

    return levels.last_message

async def get_xp_and_level(self, 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_xp_and_level(member_id: int, guild_id: int) -> tuple[float, int] async

Get both XP and level for a member.

Returns:

Type	Description
tuple[float, int]	A tuple containing (xp, level).

Source code in tux/database/controllers/levels.py

Python

async def get_xp_and_level(self, 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).
    """
    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.xp, levels.level

execute_query(query: Any) -> Any async

Execute a custom query.

Returns:

Type	Description
Any	The query result.

Source code in tux/database/controllers/levels.py

Python

    levels = await self.get_or_create_levels(member_id, guild_id)
    return levels.xp, levels.level

async def get_member_rank(self, member_id: int, guild_id: int) -> int:
    """
    Get a member's rank in their guild (1-based).

    Returns
    -------
    int

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

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

Returns:

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

Source code in tux/database/controllers/levels.py

Python

async def get_member_rank(self, 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.
    """
    levels = await self.get_levels_by_member(member_id, guild_id)
    if levels is None or levels.blacklisted:
        return -1

    # Count members with higher XP
    higher_count = await self.count(
        filters=(Levels.guild_id == guild_id) & (not Levels.blacklisted) & (Levels.xp > levels.xp),
    )
    return higher_count + 1

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

Find records using JSON column queries.

Returns:

Type	Description
list[ModelT]	List of records matching the JSON query.

Source code in tux/database/controllers/levels.py

Python

    """
    levels = await self.get_levels_by_member(member_id, guild_id)
    if levels is None or levels.blacklisted:
        return -1

    # Count members with higher XP
    higher_count = await self.count(
        filters=(Levels.guild_id == guild_id) & (not Levels.blacklisted) & (Levels.xp > levels.xp),
    )
    return higher_count + 1

async def get_guild_stats(self, guild_id: int) -> dict[str, Any]:
    """
    Get guild statistics.

    Returns

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

Get guild statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary containing total_members, blacklisted_count, and active_members.

Source code in tux/database/controllers/levels.py

Python

async def get_guild_stats(self, guild_id: int) -> dict[str, Any]:
    """
    Get guild statistics.

    Returns
    -------
    dict[str, Any]
        Dictionary containing total_members, blacklisted_count, and active_members.
    """
    total_members = await self.count(filters=Levels.guild_id == guild_id)
    blacklisted_count = await self.count(filters=(Levels.guild_id == guild_id) & (Levels.blacklisted))
    active_members = total_members - blacklisted_count

    return {
        "total_members": total_members,
        "blacklisted_count": blacklisted_count,
        "active_members": active_members,
    }

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

Find records where array column contains value.

Returns:

Type	Description
list[ModelT]	List of records with matching array values.

Source code in tux/database/controllers/levels.py

Python

dict[str, Any]
    Dictionary containing total_members, blacklisted_count, and active_members.
"""
total_members = await self.count(filters=Levels.guild_id == guild_id)
blacklisted_count = await self.count(filters=(Levels.guild_id == guild_id) & (Levels.blacklisted))
active_members = total_members - blacklisted_count

return {
    "total_members": total_members,
    "blacklisted_count": blacklisted_count,
    "active_members": active_members,
}

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

Find records using full-text search.

Returns:

Type	Description
list[ModelT]	List of records matching the search term.

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

Paginate records with metadata.

Returns:

Type	Description
PaginationResult[ModelT]	Pagination result with items, total, and page info.

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

Find paginated records with relationship loading.

Returns:

Type	Description
PaginationResult[ModelT]	Pagination result with items and relationships loaded.

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

Create multiple records in bulk.

Returns:

Type	Description
list[ModelT]	List of created records.

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

Update multiple records in bulk.

Returns:

Type	Description
int	Number of records updated.

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

Delete multiple records in bulk.

Returns:

Type	Description
int	Number of records deleted.

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

Update records matching filters.

Returns:

Type	Description
int	Number of records updated.

delete_where(filters: Any) -> int async

Delete records matching filters.

Returns:

Type	Description
int	Number of records deleted.

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

Bulk upsert with conflict resolution.

Returns:

Type	Description
list[ModelT]	List of upserted records.

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

Execute operation within a session context.

Returns:

Type	Description
R	The result of the operation.

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

Execute operation within a transaction context.

Returns:

Type	Description
R	The result of the operation.

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

Execute a callback within a transaction.

Returns:

Type	Description
Any	The result of the callback.

get_table_statistics() -> dict[str, Any] async

Get comprehensive table statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary containing table statistics.

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

Explain query performance with optional analysis.

Returns:

Type	Description
dict[str, Any]	Dictionary containing query execution plan and statistics.

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

Upsert a record by a specific field.

Returns:

Type	Description
tuple[ModelT, bool]	Tuple of (record, created) where created is True if new record was created.

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

Upsert a record by ID.

Returns:

Type	Description
tuple[ModelT, bool]	Tuple of (record, created) where created is True if new record was created.

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

Get existing record or create new one by field.

Returns:

Type	Description
tuple[ModelT, bool]	Tuple of (record, created) where created is True if new record was created.

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

Get existing record or create new one.

Returns:

Type	Description
tuple[ModelT, bool]	Tuple of (record, created) where created is True if new record was created.

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

Upsert a record.

Returns:

Type	Description
tuple[ModelT, bool]	Tuple of (record, created) where created is True if new record was created.