Skip to content

tux.database.controllers.starboard

Classes:

Name Description
StarboardController

Controller for managing starboards.
StarboardMessageController

Controller for managing starboard messages.

Classes

StarboardController()

Bases: BaseController[Starboard]

Controller for managing starboards.

This controller provides methods for creating, retrieving, updating, and deleting starboards for guilds.

Initialize the StarboardController with the starboard table.

Methods:

Name Description
get_all_starboards

Get all starboards.
get_starboard_by_guild_id

Get a starboard by guild ID.
create_or_update_starboard

Create or update a starboard.
delete_starboard_by_guild_id

Delete a starboard by guild ID.
count_starboards

Count all starboards.
find_one

Find the first matching record using SQLModel select().
find_unique

Finds a single record by a unique constraint (e.g., ID).
find_many

Finds multiple records matching specified criteria.
count

Counts records matching the specified criteria.
create

Creates a new record in the table.
update

Updates a single existing record matching the criteria.
delete

Deletes a single record matching the criteria.
upsert

Updates a record if it exists, otherwise creates it.
update_many

Updates multiple records matching the criteria.
delete_many

Deletes multiple records matching the criteria.
execute_transaction

Executes a series of database operations within a transaction.
connect_or_create_relation

Builds a SQLModel 'connect_or_create' relation structure.
safe_get_attr

Safely retrieves an attribute from an object, returning a default if absent.
Source code in tux/database/controllers/starboard.py
Python
def __init__(self):
    """Initialize the StarboardController with the starboard table."""
    super().__init__("starboard")
    self.guild_table: GuildActions[Guild] = db.client.guild

Functions

get_all_starboards() -> list[Starboard] async

Get all starboards.

Returns:

Type Description
list[Starboard]

A list of all starboards
Source code in tux/database/controllers/starboard.py
Python
async def get_all_starboards(self) -> list[Starboard]:
    """Get all starboards.

    Returns
    -------
    list[Starboard]
        A list of all starboards
    """
    return await self.find_many(where={})
get_starboard_by_guild_id(guild_id: int) -> Starboard | None async

Get a starboard by guild ID.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
Starboard | None

The starboard if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def get_starboard_by_guild_id(self, guild_id: int) -> Starboard | None:
    """Get a starboard by guild ID.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    Starboard | None
        The starboard if found, None otherwise
    """
    return await self.find_unique(where={"guild_id": guild_id})
create_or_update_starboard(guild_id: int, starboard_channel_id: int, starboard_emoji: str, starboard_threshold: int) -> Starboard async

Create or update a starboard.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required
starboard_channel_id int

The ID of the starboard channel

 required
starboard_emoji str

The emoji to use for the starboard

 required
starboard_threshold int

The threshold for the starboard

 required

Returns:

Type Description
Starboard

The created or updated starboard
Source code in tux/database/controllers/starboard.py
Python
async def create_or_update_starboard(
    self,
    guild_id: int,
    starboard_channel_id: int,
    starboard_emoji: str,
    starboard_threshold: int,
) -> Starboard:
    """Create or update a starboard.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    starboard_channel_id : int
        The ID of the starboard channel
    starboard_emoji : str
        The emoji to use for the starboard
    starboard_threshold : int
        The threshold for the starboard

    Returns
    -------
    Starboard
        The created or updated starboard
    """
    return await self.upsert(
        where={"guild_id": guild_id},
        create={
            "starboard_channel_id": starboard_channel_id,
            "starboard_emoji": starboard_emoji,
            "starboard_threshold": starboard_threshold,
            "guild_id": guild_id,
        },
        update={
            "starboard_channel_id": starboard_channel_id,
            "starboard_emoji": starboard_emoji,
            "starboard_threshold": starboard_threshold,
        },
    )
delete_starboard_by_guild_id(guild_id: int) -> Starboard | None async

Delete a starboard by guild ID.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
Starboard | None

The deleted starboard if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def delete_starboard_by_guild_id(self, guild_id: int) -> Starboard | None:
    """Delete a starboard by guild ID.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    Starboard | None
        The deleted starboard if found, None otherwise
    """
    return await self.delete(where={"guild_id": guild_id})
_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None

Adds the 'include' argument to a dictionary if it is not None.

Source code in tux/database/controllers/starboard.py
Python
Parameters
----------
guild_id : int
_build_find_args(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> dict[str, Any]

Constructs the keyword arguments dictionary for SQLModel find operations.

Source code in tux/database/controllers/starboard.py
Python
        Returns
        -------
        Starboard | None
            The deleted starboard if found, None otherwise
        """
        return await self.delete(where={"guild_id": guild_id})

    async def count_starboards(self) -> int:
        """Count all starboards.

        Returns
        -------
        int
            The number of starboards
        """
        return await self.count(where={})


class StarboardMessageController(BaseController[StarboardMessage]):
    """Controller for managing starboard messages.
count_starboards() -> int async

Count all starboards.

Returns:

Type Description
int

The number of starboards
Source code in tux/database/controllers/starboard.py
Python
async def count_starboards(self) -> int:
    """Count all starboards.

    Returns
    -------
    int
        The number of starboards
    """
    return await self.count(where={})
_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs simple keyword arguments for SQLModel (e.g., create, delete).

Source code in tux/database/controllers/starboard.py
Python
This controller provides methods for creating, retrieving, updating,
and deleting starboard messages.
"""

def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""
    super().__init__("starboardmessage")
    self.guild_table: GuildActions[Guild] = db.client.guild

async def get_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel create operations.

Source code in tux/database/controllers/starboard.py
Python
Parameters
----------
message_id : int
    The ID of the message
guild_id : int
    The ID of the guild
_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel update operations.

Source code in tux/database/controllers/starboard.py
Python
    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_unique(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )

async def create_or_update_starboard_message(
_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel delete operations.

Source code in tux/database/controllers/starboard.py
Python
message_id: int,
message_content: str,
message_expires_at: datetime,
message_channel_id: int,
message_user_id: int,
message_guild_id: int,
star_count: int,
_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel upsert operations.

Source code in tux/database/controllers/starboard.py
Python
) -> StarboardMessage:
    """Create or update a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    message_content : str
        The content of the message
    message_expires_at : datetime
        The expiration date of the message
    message_channel_id : int
        The ID of the channel the message was sent in
    message_user_id : int
        The ID of the user who sent the message
    message_guild_id : int
        The ID of the guild the message was sent in
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> ModelType | None async

Find the first matching record using SQLModel select().

Source code in tux/database/controllers/starboard.py
Python
    The ID of the starboard message

Returns
-------
StarboardMessage
    The created or updated starboard message
"""

# Use transaction to ensure atomicity of guild creation and message upsert
async def create_or_update_tx():
    # Ensure guild exists through connect_or_create in the upsert
    return await self.upsert(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": message_guild_id}},
        create={
            "message_id": message_id,
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Finds a single record by a unique constraint (e.g., ID).

Parameters:

Name Type Description Default
where dict[str, Any]

Unique query conditions (e.g., {'id': 1}).

 required
include dict[str, bool]

Specifies relations to include in the result.

 None

Returns:

Type Description
ModelType | None

The found record or None if no match exists.
Source code in tux/database/controllers/starboard.py
Python
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "message_guild_id": message_guild_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
            update={
                "message_content": message_content,
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
        )

    return await self.execute_transaction(create_or_update_tx)

async def delete_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Delete a starboard message by message ID and guild ID.
find_many(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> list[ModelType] async

Finds multiple records matching specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required
include dict[str, bool]

Specifies relations to include in the results.

 None
order dict[str, str]

Specifies the field and direction for ordering.

 None
take int

Maximum number of records to return.

 None
skip int

Number of records to skip (for pagination).

 None
cursor dict[str, Any]

Cursor for pagination based on a unique field.

 None

Returns:

Type Description
list[ModelType]

A list of found records, potentially empty.
Source code in tux/database/controllers/starboard.py
Python
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The deleted starboard message if found, None otherwise
    """
    return await self.delete(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )

async def get_all_starboard_messages(
    self,
    guild_id: int,
    limit: int | None = None,
    order_by_stars: bool = False,
) -> list[StarboardMessage]:
    """Get all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int | None
        Optional limit on the number of messages to return
    order_by_stars : bool
        Whether to order by star count (highest first)

    Returns
    -------
    list[StarboardMessage]
        A list of all starboard messages for the guild
    """
    order = {"star_count": "desc"} if order_by_stars else {"message_expires_at": "desc"}

    return await self.find_many(
        where={"message_guild_id": guild_id},
        order=order,
count(where: dict[str, Any]) -> int async

Counts records matching the specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required

Returns:

Type Description
int

The total number of matching records.
Source code in tux/database/controllers/starboard.py
Python
    )

async def update_star_count(self, message_id: int, guild_id: int, new_star_count: int) -> StarboardMessage | None:
    """Update the star count of a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild
    new_star_count : int
        The new star count

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType async

Creates a new record in the table.

Parameters:

Name Type Description Default
data dict[str, Any]

The data for the new record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The newly created record.
Source code in tux/database/controllers/starboard.py
Python
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
        data={"star_count": new_star_count},
    )

async def get_starboard_message_by_id(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Get a starboard message by its ID and guild ID.

    A "starboard message" is the response by the bot, not the original message.

    Parameters
    ----------
    message_id : int
        The ID of the starboard message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_one(where={"message_id": message_id, "message_guild_id": guild_id})

async def increment_star_count(self, message_id: int, guild_id: int) -> StarboardMessage | None:
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Updates a single existing record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to update.

 required
data dict[str, Any]

The data to update the record with.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType | None

The updated record, or None if no matching record was found.
Source code in tux/database/controllers/starboard.py
Python
    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """

    async def increment_tx():
        message = await self.get_starboard_message(message_id, guild_id)
        if message is None:
            return None

        star_count = self.safe_get_attr(message, "star_count", 0)
        return await self.update_star_count(message_id, guild_id, star_count + 1)

    return await self.execute_transaction(increment_tx)

async def get_top_starred_messages(self, guild_id: int, limit: int = 10) -> list[StarboardMessage]:
    """Get the top starred messages for a guild.

    Parameters
    ----------
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Deletes a single record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to delete.

 required
include dict[str, bool]

Specifies relations to include in the returned deleted record.

 None

Returns:

Type Description
ModelType | None

The deleted record, or None if no matching record was found.
Source code in tux/database/controllers/starboard.py
Python
        The ID of the guild
    limit : int
        The maximum number of messages to return

    Returns
    -------
    list[StarboardMessage]
        The top starred messages
    """
    return await self.find_many(
        where={"message_guild_id": guild_id},
        order={"star_count": "desc"},
        take=limit,
    )

async def count_starboard_messages(self, guild_id: int) -> int:
    """Count the number of starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType async

Updates a record if it exists, otherwise creates it.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the existing record.

 required
create dict[str, Any]

Data to use if creating a new record.

 required
update dict[str, Any]

Data to use if updating an existing record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The created or updated record.
Source code in tux/database/controllers/starboard.py
Python
        The number of starboard messages
    """
    return await self.count(where={"message_guild_id": guild_id})

async def bulk_delete_messages_by_guild_id(self, guild_id: int) -> int:
    """Delete all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of messages deleted
    """
    return await self.delete_many(where={"message_guild_id": guild_id})

async def get_messages_for_user(
    self,
    user_id: int,
    guild_id: int | None = None,
    limit: int | None = None,
) -> list[StarboardMessage]:
    """Get all starboard messages for a user.

    Parameters
    ----------
    user_id : int
        The ID of the user
    guild_id : int | None
        Optional guild ID to filter by
    limit : int | None
        Optional limit on the number of messages to return
update_many(where: dict[str, Any], data: dict[str, Any]) -> int async

Updates multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to update.

 required
data dict[str, Any]

The data to update the records with.

 required

Returns:

Type Description
int

The number of records updated.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
Source code in tux/database/controllers/starboard.py
Python
Returns
-------
list[StarboardMessage]
    The starboard messages for the user
"""
where = {"message_user_id": user_id}
if guild_id is not None:
    where["message_guild_id"] = guild_id

return await self.find_many(
    where=where,
    order={"star_count": "desc"},
    take=limit,
)
delete_many(where: dict[str, Any]) -> int async

Deletes multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to delete.

 required

Returns:

Type Description
int

The number of records deleted.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
execute_transaction(callback: Callable[[], Any]) -> Any async

Executes a series of database operations within a transaction.

Ensures atomicity: all operations succeed or all fail and roll back. Note: Does not use _execute_query internally to preserve specific transaction context in error messages.

Parameters:

Name Type Description Default
callback Callable[[], Any]

An async function containing the database operations to execute.

 required

Returns:

Type Description
Any

The result returned by the callback function.

Raises:

Type Description
Exception

Re-raises any exception that occurs during the transaction.
connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod

Builds a SQLModel 'connect_or_create' relation structure.

Simplifies linking or creating related records during create/update operations.

Parameters:

Name Type Description Default
id_field str

The name of the ID field used for connection (e.g., 'guild_id').

 required
model_id Any

The ID value of the record to connect to.

 required
create_data dict[str, Any]

Additional data required if creating the related record. Must include at least the id_field and model_id.

 None

Returns:

Type Description
dict[str, Any]

A dictionary formatted for SQLModel's connect_or_create.
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Safely retrieves an attribute from an object, returning a default if absent.

Parameters:

Name Type Description Default
obj Any

The object to retrieve the attribute from.

 required
attr str

The name of the attribute.

 required
default Any

The value to return if the attribute is not found. Defaults to None.

 None

Returns:

Type Description
Any

The attribute's value or the default value.

StarboardMessageController()

Bases: BaseController[StarboardMessage]

Controller for managing starboard messages.

This controller provides methods for creating, retrieving, updating, and deleting starboard messages.

Initialize the StarboardMessageController with the starboardmessage table.

Methods:

Name Description
get_starboard_message

Get a starboard message by message ID and guild ID.
create_or_update_starboard_message

Create or update a starboard message.
find_one

Find the first matching record using SQLModel select().
find_unique

Finds a single record by a unique constraint (e.g., ID).
delete_starboard_message

Delete a starboard message by message ID and guild ID.
find_many

Finds multiple records matching specified criteria.
get_all_starboard_messages

Get all starboard messages for a guild.
count

Counts records matching the specified criteria.
update_star_count

Update the star count of a starboard message.
create

Creates a new record in the table.
get_starboard_message_by_id

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

Increment the star count of a starboard message.
update

Updates a single existing record matching the criteria.
get_top_starred_messages

Get the top starred messages for a guild.
delete

Deletes a single record matching the criteria.
count_starboard_messages

Count the number of starboard messages for a guild.
upsert

Updates a record if it exists, otherwise creates it.
bulk_delete_messages_by_guild_id

Delete all starboard messages for a guild.
get_messages_for_user

Get all starboard messages for a user.
update_many

Updates multiple records matching the criteria.
delete_many

Deletes multiple records matching the criteria.
execute_transaction

Executes a series of database operations within a transaction.
connect_or_create_relation

Builds a SQLModel 'connect_or_create' relation structure.
safe_get_attr

Safely retrieves an attribute from an object, returning a default if absent.
Source code in tux/database/controllers/starboard.py
Python
def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""
    super().__init__("starboardmessage")
    self.guild_table: GuildActions[Guild] = db.client.guild

Functions

_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None

Adds the 'include' argument to a dictionary if it is not None.

Source code in tux/database/controllers/starboard.py
Python
Parameters
----------
guild_id : int
_build_find_args(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> dict[str, Any]

Constructs the keyword arguments dictionary for SQLModel find operations.

Source code in tux/database/controllers/starboard.py
Python
        Returns
        -------
        Starboard | None
            The deleted starboard if found, None otherwise
        """
        return await self.delete(where={"guild_id": guild_id})

    async def count_starboards(self) -> int:
        """Count all starboards.

        Returns
        -------
        int
            The number of starboards
        """
        return await self.count(where={})


class StarboardMessageController(BaseController[StarboardMessage]):
    """Controller for managing starboard messages.
_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs simple keyword arguments for SQLModel (e.g., create, delete).

Source code in tux/database/controllers/starboard.py
Python
This controller provides methods for creating, retrieving, updating,
and deleting starboard messages.
"""

def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""
    super().__init__("starboardmessage")
    self.guild_table: GuildActions[Guild] = db.client.guild

async def get_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
get_starboard_message(message_id: int, guild_id: int) -> StarboardMessage | None async

Get a starboard message by message ID and guild ID.

Parameters:

Name Type Description Default
message_id int

The ID of the message

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
StarboardMessage | None

The starboard message if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def get_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Get a starboard message by message ID and guild ID.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_unique(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel create operations.

Source code in tux/database/controllers/starboard.py
Python
Parameters
----------
message_id : int
    The ID of the message
guild_id : int
    The ID of the guild
_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel update operations.

Source code in tux/database/controllers/starboard.py
Python
    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_unique(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )

async def create_or_update_starboard_message(
create_or_update_starboard_message(message_id: int, message_content: str, message_expires_at: datetime, message_channel_id: int, message_user_id: int, message_guild_id: int, star_count: int, starboard_message_id: int) -> StarboardMessage async

Create or update a starboard message.

Parameters:

Name Type Description Default
message_id int

The ID of the message

 required
message_content str

The content of the message

 required
message_expires_at datetime

The expiration date of the message

 required
message_channel_id int

The ID of the channel the message was sent in

 required
message_user_id int

The ID of the user who sent the message

 required
message_guild_id int

The ID of the guild the message was sent in

 required
star_count int

The number of stars the message has

 required
starboard_message_id int

The ID of the starboard message

 required

Returns:

Type Description
StarboardMessage

The created or updated starboard message
Source code in tux/database/controllers/starboard.py
Python
async def create_or_update_starboard_message(
    self,
    message_id: int,
    message_content: str,
    message_expires_at: datetime,
    message_channel_id: int,
    message_user_id: int,
    message_guild_id: int,
    star_count: int,
    starboard_message_id: int,
) -> StarboardMessage:
    """Create or update a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    message_content : str
        The content of the message
    message_expires_at : datetime
        The expiration date of the message
    message_channel_id : int
        The ID of the channel the message was sent in
    message_user_id : int
        The ID of the user who sent the message
    message_guild_id : int
        The ID of the guild the message was sent in
    star_count : int
        The number of stars the message has
    starboard_message_id : int
        The ID of the starboard message

    Returns
    -------
    StarboardMessage
        The created or updated starboard message
    """

    # Use transaction to ensure atomicity of guild creation and message upsert
    async def create_or_update_tx():
        # Ensure guild exists through connect_or_create in the upsert
        return await self.upsert(
            where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": message_guild_id}},
            create={
                "message_id": message_id,
                "message_content": message_content,
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "message_guild_id": message_guild_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
            update={
                "message_content": message_content,
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
        )

    return await self.execute_transaction(create_or_update_tx)
_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel delete operations.

Source code in tux/database/controllers/starboard.py
Python
message_id: int,
message_content: str,
message_expires_at: datetime,
message_channel_id: int,
message_user_id: int,
message_guild_id: int,
star_count: int,
_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel upsert operations.

Source code in tux/database/controllers/starboard.py
Python
) -> StarboardMessage:
    """Create or update a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    message_content : str
        The content of the message
    message_expires_at : datetime
        The expiration date of the message
    message_channel_id : int
        The ID of the channel the message was sent in
    message_user_id : int
        The ID of the user who sent the message
    message_guild_id : int
        The ID of the guild the message was sent in
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> ModelType | None async

Find the first matching record using SQLModel select().

Source code in tux/database/controllers/starboard.py
Python
    The ID of the starboard message

Returns
-------
StarboardMessage
    The created or updated starboard message
"""

# Use transaction to ensure atomicity of guild creation and message upsert
async def create_or_update_tx():
    # Ensure guild exists through connect_or_create in the upsert
    return await self.upsert(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": message_guild_id}},
        create={
            "message_id": message_id,
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Finds a single record by a unique constraint (e.g., ID).

Parameters:

Name Type Description Default
where dict[str, Any]

Unique query conditions (e.g., {'id': 1}).

 required
include dict[str, bool]

Specifies relations to include in the result.

 None

Returns:

Type Description
ModelType | None

The found record or None if no match exists.
Source code in tux/database/controllers/starboard.py
Python
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "message_guild_id": message_guild_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
            update={
                "message_content": message_content,
                "message_expires_at": message_expires_at,
                "message_channel_id": message_channel_id,
                "message_user_id": message_user_id,
                "star_count": star_count,
                "starboard_message_id": starboard_message_id,
            },
        )

    return await self.execute_transaction(create_or_update_tx)

async def delete_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Delete a starboard message by message ID and guild ID.
delete_starboard_message(message_id: int, guild_id: int) -> StarboardMessage | None async

Delete a starboard message by message ID and guild ID.

Parameters:

Name Type Description Default
message_id int

The ID of the message

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
StarboardMessage | None

The deleted starboard message if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def delete_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Delete a starboard message by message ID and guild ID.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The deleted starboard message if found, None otherwise
    """
    return await self.delete(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )
find_many(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None, cursor: dict[str, Any] | None = None) -> list[ModelType] async

Finds multiple records matching specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required
include dict[str, bool]

Specifies relations to include in the results.

 None
order dict[str, str]

Specifies the field and direction for ordering.

 None
take int

Maximum number of records to return.

 None
skip int

Number of records to skip (for pagination).

 None
cursor dict[str, Any]

Cursor for pagination based on a unique field.

 None

Returns:

Type Description
list[ModelType]

A list of found records, potentially empty.
Source code in tux/database/controllers/starboard.py
Python
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The deleted starboard message if found, None otherwise
    """
    return await self.delete(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
    )

async def get_all_starboard_messages(
    self,
    guild_id: int,
    limit: int | None = None,
    order_by_stars: bool = False,
) -> list[StarboardMessage]:
    """Get all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int | None
        Optional limit on the number of messages to return
    order_by_stars : bool
        Whether to order by star count (highest first)

    Returns
    -------
    list[StarboardMessage]
        A list of all starboard messages for the guild
    """
    order = {"star_count": "desc"} if order_by_stars else {"message_expires_at": "desc"}

    return await self.find_many(
        where={"message_guild_id": guild_id},
        order=order,
get_all_starboard_messages(guild_id: int, limit: int | None = None, order_by_stars: bool = False) -> list[StarboardMessage] async

Get all starboard messages for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required
limit int | None

Optional limit on the number of messages to return

 None
order_by_stars bool

Whether to order by star count (highest first)

 False

Returns:

Type Description
list[StarboardMessage]

A list of all starboard messages for the guild
Source code in tux/database/controllers/starboard.py
Python
async def get_all_starboard_messages(
    self,
    guild_id: int,
    limit: int | None = None,
    order_by_stars: bool = False,
) -> list[StarboardMessage]:
    """Get all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int | None
        Optional limit on the number of messages to return
    order_by_stars : bool
        Whether to order by star count (highest first)

    Returns
    -------
    list[StarboardMessage]
        A list of all starboard messages for the guild
    """
    order = {"star_count": "desc"} if order_by_stars else {"message_expires_at": "desc"}

    return await self.find_many(
        where={"message_guild_id": guild_id},
        order=order,
        take=limit,
    )
count(where: dict[str, Any]) -> int async

Counts records matching the specified criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to match.

 required

Returns:

Type Description
int

The total number of matching records.
Source code in tux/database/controllers/starboard.py
Python
    )

async def update_star_count(self, message_id: int, guild_id: int, new_star_count: int) -> StarboardMessage | None:
    """Update the star count of a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild
    new_star_count : int
        The new star count

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """
update_star_count(message_id: int, guild_id: int, new_star_count: int) -> StarboardMessage | None async

Update the star count of a starboard message.

Parameters:

Name Type Description Default
message_id int

The ID of the message

 required
guild_id int

The ID of the guild

 required
new_star_count int

The new star count

 required

Returns:

Type Description
StarboardMessage | None

The updated starboard message if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def update_star_count(self, message_id: int, guild_id: int, new_star_count: int) -> StarboardMessage | None:
    """Update the star count of a starboard message.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild
    new_star_count : int
        The new star count

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """
    return await self.update(
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
        data={"star_count": new_star_count},
    )
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType async

Creates a new record in the table.

Parameters:

Name Type Description Default
data dict[str, Any]

The data for the new record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The newly created record.
Source code in tux/database/controllers/starboard.py
Python
        where={"message_id_message_guild_id": {"message_id": message_id, "message_guild_id": guild_id}},
        data={"star_count": new_star_count},
    )

async def get_starboard_message_by_id(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Get a starboard message by its ID and guild ID.

    A "starboard message" is the response by the bot, not the original message.

    Parameters
    ----------
    message_id : int
        The ID of the starboard message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_one(where={"message_id": message_id, "message_guild_id": guild_id})

async def increment_star_count(self, message_id: int, guild_id: int) -> StarboardMessage | None:
get_starboard_message_by_id(message_id: int, guild_id: int) -> StarboardMessage | None async

Get a starboard message by its ID and guild ID.

A "starboard message" is the response by the bot, not the original message.

Parameters:

Name Type Description Default
message_id int

The ID of the starboard message

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
StarboardMessage | None

The starboard message if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def get_starboard_message_by_id(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Get a starboard message by its ID and guild ID.

    A "starboard message" is the response by the bot, not the original message.

    Parameters
    ----------
    message_id : int
        The ID of the starboard message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The starboard message if found, None otherwise
    """
    return await self.find_one(where={"message_id": message_id, "message_guild_id": guild_id})
increment_star_count(message_id: int, guild_id: int) -> StarboardMessage | None async

Increment the star count of a starboard message.

This method uses a transaction to ensure atomicity.

Parameters:

Name Type Description Default
message_id int

The ID of the message

 required
guild_id int

The ID of the guild

 required

Returns:

Type Description
StarboardMessage | None

The updated starboard message if found, None otherwise
Source code in tux/database/controllers/starboard.py
Python
async def increment_star_count(self, message_id: int, guild_id: int) -> StarboardMessage | None:
    """Increment the star count of a starboard message.

    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """

    async def increment_tx():
        message = await self.get_starboard_message(message_id, guild_id)
        if message is None:
            return None

        star_count = self.safe_get_attr(message, "star_count", 0)
        return await self.update_star_count(message_id, guild_id, star_count + 1)

    return await self.execute_transaction(increment_tx)
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Updates a single existing record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to update.

 required
data dict[str, Any]

The data to update the record with.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType | None

The updated record, or None if no matching record was found.
Source code in tux/database/controllers/starboard.py
Python
    This method uses a transaction to ensure atomicity.

    Parameters
    ----------
    message_id : int
        The ID of the message
    guild_id : int
        The ID of the guild

    Returns
    -------
    StarboardMessage | None
        The updated starboard message if found, None otherwise
    """

    async def increment_tx():
        message = await self.get_starboard_message(message_id, guild_id)
        if message is None:
            return None

        star_count = self.safe_get_attr(message, "star_count", 0)
        return await self.update_star_count(message_id, guild_id, star_count + 1)

    return await self.execute_transaction(increment_tx)

async def get_top_starred_messages(self, guild_id: int, limit: int = 10) -> list[StarboardMessage]:
    """Get the top starred messages for a guild.

    Parameters
    ----------
get_top_starred_messages(guild_id: int, limit: int = 10) -> list[StarboardMessage] async

Get the top starred messages for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required
limit int

The maximum number of messages to return

 10

Returns:

Type Description
list[StarboardMessage]

The top starred messages
Source code in tux/database/controllers/starboard.py
Python
async def get_top_starred_messages(self, guild_id: int, limit: int = 10) -> list[StarboardMessage]:
    """Get the top starred messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild
    limit : int
        The maximum number of messages to return

    Returns
    -------
    list[StarboardMessage]
        The top starred messages
    """
    return await self.find_many(
        where={"message_guild_id": guild_id},
        order={"star_count": "desc"},
        take=limit,
    )
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | None async

Deletes a single record matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the record to delete.

 required
include dict[str, bool]

Specifies relations to include in the returned deleted record.

 None

Returns:

Type Description
ModelType | None

The deleted record, or None if no matching record was found.
Source code in tux/database/controllers/starboard.py
Python
        The ID of the guild
    limit : int
        The maximum number of messages to return

    Returns
    -------
    list[StarboardMessage]
        The top starred messages
    """
    return await self.find_many(
        where={"message_guild_id": guild_id},
        order={"star_count": "desc"},
        take=limit,
    )

async def count_starboard_messages(self, guild_id: int) -> int:
    """Count the number of starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
count_starboard_messages(guild_id: int) -> int async

Count the number of starboard messages for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

The number of starboard messages
Source code in tux/database/controllers/starboard.py
Python
async def count_starboard_messages(self, guild_id: int) -> int:
    """Count the number of starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of starboard messages
    """
    return await self.count(where={"message_guild_id": guild_id})
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType async

Updates a record if it exists, otherwise creates it.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the existing record.

 required
create dict[str, Any]

Data to use if creating a new record.

 required
update dict[str, Any]

Data to use if updating an existing record.

 required
include dict[str, bool]

Specifies relations to include in the returned record.

 None

Returns:

Type Description
ModelType

The created or updated record.
Source code in tux/database/controllers/starboard.py
Python
        The number of starboard messages
    """
    return await self.count(where={"message_guild_id": guild_id})

async def bulk_delete_messages_by_guild_id(self, guild_id: int) -> int:
    """Delete all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of messages deleted
    """
    return await self.delete_many(where={"message_guild_id": guild_id})

async def get_messages_for_user(
    self,
    user_id: int,
    guild_id: int | None = None,
    limit: int | None = None,
) -> list[StarboardMessage]:
    """Get all starboard messages for a user.

    Parameters
    ----------
    user_id : int
        The ID of the user
    guild_id : int | None
        Optional guild ID to filter by
    limit : int | None
        Optional limit on the number of messages to return
bulk_delete_messages_by_guild_id(guild_id: int) -> int async

Delete all starboard messages for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild

 required

Returns:

Type Description
int

The number of messages deleted
Source code in tux/database/controllers/starboard.py
Python
async def bulk_delete_messages_by_guild_id(self, guild_id: int) -> int:
    """Delete all starboard messages for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild

    Returns
    -------
    int
        The number of messages deleted
    """
    return await self.delete_many(where={"message_guild_id": guild_id})
get_messages_for_user(user_id: int, guild_id: int | None = None, limit: int | None = None) -> list[StarboardMessage] async

Get all starboard messages for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user

 required
guild_id int | None

Optional guild ID to filter by

 None
limit int | None

Optional limit on the number of messages to return

 None

Returns:

Type Description
list[StarboardMessage]

The starboard messages for the user
Source code in tux/database/controllers/starboard.py
Python
async def get_messages_for_user(
    self,
    user_id: int,
    guild_id: int | None = None,
    limit: int | None = None,
) -> list[StarboardMessage]:
    """Get all starboard messages for a user.

    Parameters
    ----------
    user_id : int
        The ID of the user
    guild_id : int | None
        Optional guild ID to filter by
    limit : int | None
        Optional limit on the number of messages to return

    Returns
    -------
    list[StarboardMessage]
        The starboard messages for the user
    """
    where = {"message_user_id": user_id}
    if guild_id is not None:
        where["message_guild_id"] = guild_id

    return await self.find_many(
        where=where,
        order={"star_count": "desc"},
        take=limit,
    )
update_many(where: dict[str, Any], data: dict[str, Any]) -> int async

Updates multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to update.

 required
data dict[str, Any]

The data to update the records with.

 required

Returns:

Type Description
int

The number of records updated.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
Source code in tux/database/controllers/starboard.py
Python
Returns
-------
list[StarboardMessage]
    The starboard messages for the user
"""
where = {"message_user_id": user_id}
if guild_id is not None:
    where["message_guild_id"] = guild_id

return await self.find_many(
    where=where,
    order={"star_count": "desc"},
    take=limit,
)
delete_many(where: dict[str, Any]) -> int async

Deletes multiple records matching the criteria.

Parameters:

Name Type Description Default
where dict[str, Any]

Query conditions to find the records to delete.

 required

Returns:

Type Description
int

The number of records deleted.

Raises:

Type Description
ValueError

If the database operation does not return a valid count.
execute_transaction(callback: Callable[[], Any]) -> Any async

Executes a series of database operations within a transaction.

Ensures atomicity: all operations succeed or all fail and roll back. Note: Does not use _execute_query internally to preserve specific transaction context in error messages.

Parameters:

Name Type Description Default
callback Callable[[], Any]

An async function containing the database operations to execute.

 required

Returns:

Type Description
Any

The result returned by the callback function.

Raises:

Type Description
Exception

Re-raises any exception that occurs during the transaction.
connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod

Builds a SQLModel 'connect_or_create' relation structure.

Simplifies linking or creating related records during create/update operations.

Parameters:

Name Type Description Default
id_field str

The name of the ID field used for connection (e.g., 'guild_id').

 required
model_id Any

The ID value of the record to connect to.

 required
create_data dict[str, Any]

Additional data required if creating the related record. Must include at least the id_field and model_id.

 None

Returns:

Type Description
dict[str, Any]

A dictionary formatted for SQLModel's connect_or_create.
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Safely retrieves an attribute from an object, returning a default if absent.

Parameters:

Name Type Description Default
obj Any

The object to retrieve the attribute from.

 required
attr str

The name of the attribute.

 required
default Any

The value to return if the attribute is not found. Defaults to None.

 None

Returns:

Type Description
Any

The attribute's value or the default value.