`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`	Finds the first record matching specified criteria.
`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 Prisma '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,
        },
    )

`_execute_query(operation: Callable[[], Any], error_msg: str) -> Any` `async` ¶

Executes a database query with standardized error logging.

Wraps the Prisma client operation call in a try-except block, logging any exceptions with a contextual error message.

Parameters:

Name	Type	Description	Default
`operation`	`Callable[[], Any]`	A zero-argument function (e.g., a lambda) that performs the database call.	required
`error_msg`	`str`	The base error message to log if an exception occurs.	required

Returns:

Type	Description
`Any`	The result of the database operation.

Raises:

Type	Description
`Exception`	Re-raises any exception caught during the database operation.

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

Python

        """
        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,
            },
        )

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

    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.

    This controller provides methods for creating, retrieving, updating,

`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})

`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={})

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

"""

def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""

`_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 Prisma find operations.

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

Python

    self.guild_table: GuildActions[Guild] = db.client.guild

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_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

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

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

Python

    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:

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma create operations.

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

Python

Parameters
----------
message_id : int
    The ID of the message
message_content : str
    The content of the message

`_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma update operations.

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

Python

    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

`_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma delete operations.

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

Python

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

`_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 Prisma upsert operations.

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

Python

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,

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Starboard | None` `async` ¶

Finds the first record 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 result.	`None`
`order`	`dict[str, str]`	Specifies the field and direction for ordering.	`None`

Returns:

Type	Description
`ModelType \| None`	The found record or None if no match exists.

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

Python

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

    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}},
    )

async def get_all_starboard_messages(
    self,

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | 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

    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,

`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[Starboard]` `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

    )

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},
    )

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

`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

"""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:

`create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard` `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

        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
    ----------
    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"},

`update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | 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

    )

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

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

`delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | 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 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

    Returns
    -------
    list[StarboardMessage]

`upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard` `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

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

`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 Prisma '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 Prisma'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`	Finds the first record matching specified criteria.
`delete_starboard_message`	Delete a starboard message by message ID and guild ID.
`get_all_starboard_messages`	Get all starboard messages for a guild.
`find_unique`	Finds a single record by a unique constraint (e.g., ID).
`find_many`	Finds multiple records matching specified criteria.
`update_star_count`	Update the star count of a starboard message.
`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.
`count`	Counts records matching the specified criteria.
`create`	Creates a new record in the table.
`get_top_starred_messages`	Get the top starred messages for a guild.
`update`	Updates a single existing record matching the criteria.
`count_starboard_messages`	Count the number of starboard messages for a guild.
`bulk_delete_messages_by_guild_id`	Delete all starboard messages for a guild.
`delete`	Deletes a single record matching the criteria.
`get_messages_for_user`	Get all starboard messages for a user.
`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 Prisma '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¶

`_execute_query(operation: Callable[[], Any], error_msg: str) -> Any` `async` ¶

Executes a database query with standardized error logging.

Wraps the Prisma client operation call in a try-except block, logging any exceptions with a contextual error message.

Parameters:

Name	Type	Description	Default
`operation`	`Callable[[], Any]`	A zero-argument function (e.g., a lambda) that performs the database call.	required
`error_msg`	`str`	The base error message to log if an exception occurs.	required

Returns:

Type	Description
`Any`	The result of the database operation.

Raises:

Type	Description
`Exception`	Re-raises any exception caught during the database operation.

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

Python

        """
        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,
            },
        )

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

    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.

    This controller provides methods for creating, retrieving, updating,

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

"""

def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""

`_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 Prisma find operations.

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

Python

    self.guild_table: GuildActions[Guild] = db.client.guild

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}},
    )

`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}},
    )

`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_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

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

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

Python

    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:

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma create operations.

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

Python

Parameters
----------
message_id : int
    The ID of the message
message_content : str
    The content of the message

`_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma update operations.

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

Python

    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

`_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma delete operations.

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

Python

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

`_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 Prisma upsert operations.

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

Python

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,

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> StarboardMessage | None` `async` ¶

Finds the first record 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 result.	`None`
`order`	`dict[str, str]`	Specifies the field and direction for ordering.	`None`

Returns:

Type	Description
`ModelType \| None`	The found record or None if no match exists.

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

Python

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

    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}},
    )

async def get_all_starboard_messages(
    self,

`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}},
    )

`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,
    )

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | 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

    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,

`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[StarboardMessage]` `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

    )

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},
    )

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

`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},
    )

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

`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

"""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:

`create(data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage` `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

        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
    ----------
    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"},

`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,
    )

`update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | 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

    )

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

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

`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})

`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})

`delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | 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 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

    Returns
    -------
    list[StarboardMessage]

`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,
    )

`upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage` `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

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

`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 Prisma '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 Prisma'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.

tux.database.controllers.starboard ¶

Classes¶

StarboardController() ¶

Functions¶

get_all_starboards() -> list[Starboard] async ¶

get_starboard_by_guild_id(guild_id: int) -> Starboard | None async ¶

create_or_update_starboard(guild_id: int, starboard_channel_id: int, starboard_emoji: str, starboard_threshold: int) -> Starboard async ¶

_execute_query(operation: Callable[[], Any], error_msg: str) -> Any async ¶

delete_starboard_by_guild_id(guild_id: int) -> Starboard | None async ¶

count_starboards() -> int async ¶

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

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

_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

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

_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Starboard | None async ¶

find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | None async ¶

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[Starboard] async ¶

count(where: dict[str, Any]) -> int async ¶

create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard async ¶

update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | None async ¶

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

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

update_many(where: dict[str, Any], data: dict[str, Any]) -> int async ¶

delete_many(where: dict[str, Any]) -> int async ¶

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

connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod ¶

safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod ¶

StarboardMessageController() ¶

Functions¶

_execute_query(operation: Callable[[], Any], error_msg: str) -> Any async ¶

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

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

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

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 ¶

_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

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

_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> StarboardMessage | None async ¶

delete_starboard_message(message_id: int, guild_id: int) -> StarboardMessage | None async ¶

get_all_starboard_messages(guild_id: int, limit: int | None = None, order_by_stars: bool = False) -> list[StarboardMessage] async ¶

find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | None async ¶

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[StarboardMessage] async ¶

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

get_starboard_message_by_id(message_id: int, guild_id: int) -> StarboardMessage | None async ¶

increment_star_count(message_id: int, guild_id: int) -> StarboardMessage | None async ¶

count(where: dict[str, Any]) -> int async ¶

create(data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage async ¶

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

update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | None async ¶

count_starboard_messages(guild_id: int) -> int async ¶

bulk_delete_messages_by_guild_id(guild_id: int) -> int async ¶

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

get_messages_for_user(user_id: int, guild_id: int | None = None, limit: int | None = None) -> list[StarboardMessage] async ¶

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

update_many(where: dict[str, Any], data: dict[str, Any]) -> int async ¶

delete_many(where: dict[str, Any]) -> int async ¶

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

connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod ¶

safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod ¶

`tux.database.controllers.starboard` ¶

`StarboardController()` ¶

`get_all_starboards() -> list[Starboard]` `async` ¶

`get_starboard_by_guild_id(guild_id: int) -> Starboard | None` `async` ¶

`create_or_update_starboard(guild_id: int, starboard_channel_id: int, starboard_emoji: str, starboard_threshold: int) -> Starboard` `async` ¶

`_execute_query(operation: Callable[[], Any], error_msg: str) -> Any` `async` ¶

`delete_starboard_by_guild_id(guild_id: int) -> Starboard | None` `async` ¶

`count_starboards() -> int` `async` ¶

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

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

`_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Starboard | None` `async` ¶

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | None` `async` ¶

`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[Starboard]` `async` ¶

`count(where: dict[str, Any]) -> int` `async` ¶

`create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard` `async` ¶

`update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | None` `async` ¶

`delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Starboard | None` `async` ¶

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

`update_many(where: dict[str, Any], data: dict[str, Any]) -> int` `async` ¶

`delete_many(where: dict[str, Any]) -> int` `async` ¶

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

`connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any]` `staticmethod` ¶

`safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any` `staticmethod` ¶

`StarboardMessageController()` ¶

`_execute_query(operation: Callable[[], Any], error_msg: str) -> Any` `async` ¶

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

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

`get_starboard_message(message_id: int, guild_id: int) -> StarboardMessage | None` `async` ¶

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

`_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_update_args(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> StarboardMessage | None` `async` ¶

`delete_starboard_message(message_id: int, guild_id: int) -> StarboardMessage | None` `async` ¶

`get_all_starboard_messages(guild_id: int, limit: int | None = None, order_by_stars: bool = False) -> list[StarboardMessage]` `async` ¶

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | None` `async` ¶

`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[StarboardMessage]` `async` ¶

`update_star_count(message_id: int, guild_id: int, new_star_count: int) -> StarboardMessage | None` `async` ¶

`get_starboard_message_by_id(message_id: int, guild_id: int) -> StarboardMessage | None` `async` ¶

`increment_star_count(message_id: int, guild_id: int) -> StarboardMessage | None` `async` ¶

`count(where: dict[str, Any]) -> int` `async` ¶

`create(data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage` `async` ¶

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

`update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | None` `async` ¶

`count_starboard_messages(guild_id: int) -> int` `async` ¶

`bulk_delete_messages_by_guild_id(guild_id: int) -> int` `async` ¶

`delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> StarboardMessage | None` `async` ¶

`get_messages_for_user(user_id: int, guild_id: int | None = None, limit: int | None = None) -> list[StarboardMessage]` `async` ¶

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

`update_many(where: dict[str, Any], data: dict[str, Any]) -> int` `async` ¶

`delete_many(where: dict[str, Any]) -> int` `async` ¶

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

`connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any]` `staticmethod` ¶

`safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any` `staticmethod` ¶