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

Return the first row that matches where or None.
delete_starboard_by_guild_id

Delete a starboard by guild ID.
find_many

Return a list of rows matching where (or all rows).
count_starboards

Count all starboards.
execute_transaction

Execute callback inside a database session / transaction block.
safe_get_attr

Return getattr(obj, attr, default) - keeps old helper available.
connect_or_create_relation

Return a dict with a single key that can be merged into data dicts.
Source code in tux/database/controllers/starboard.py
Python
def __init__(self):
    """Initialize the StarboardController with the starboard table."""
    super().__init__(Starboard)

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})
_execute_query(op: Callable[[AsyncSession], Any], span_desc: str) -> Any async

Run op inside a managed session & sentry span (if enabled).

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
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,
        },
    )
find_one(*, where: dict[str, Any], include: dict[str, bool] | None = None, **__: Any) -> ModelT | None async

Return the first row that matches where or None.

Source code in tux/database/controllers/starboard.py
Python
    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})
find_many(*, where: dict[str, Any] | None = None, include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None) -> list[ModelT] async

Return a list of rows matching where (or all rows).

Source code in tux/database/controllers/starboard.py
Python
            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,
    and deleting starboard messages.
    """

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

    async def get_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
        """Get a starboard message by message ID and 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={})
execute_transaction(callback: Callable[[], Any]) -> Any async

Execute callback inside a database session / transaction block.

Source code in tux/database/controllers/starboard.py
Python
    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]:
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Return getattr(obj, attr, default) - keeps old helper available.

Source code in tux/database/controllers/starboard.py
Python
    The ID of the guild
limit : int | None
    Optional limit on the number of messages to return
order_by_stars : bool
connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any] staticmethod

Return a dict with a single key that can be merged into data dicts.

The calling code does something like::

Text Only
data = {"guild": connect_or_create_relation("guild_id", guild_id)}

We map that pattern to a very small helper that collapses to {"guild_id": guild_id}.

Source code in tux/database/controllers/starboard.py
Python
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,
)

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
find_one

Return the first row that matches where or None.
find_many

Return a list of rows matching where (or all rows).
get_starboard_message

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

Create or update a starboard message.
delete_starboard_message

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

Execute callback inside a database session / transaction block.
get_all_starboard_messages

Get all starboard messages for a guild.
safe_get_attr

Return getattr(obj, attr, default) - keeps old helper available.
connect_or_create_relation

Return a dict with a single key that can be merged into data dicts.
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.
get_top_starred_messages

Get the top starred messages for a guild.
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.
get_messages_for_user

Get all starboard messages for a user.
Source code in tux/database/controllers/starboard.py
Python
def __init__(self):
    """Initialize the StarboardMessageController with the starboardmessage table."""
    super().__init__(StarboardMessage)

Functions

_execute_query(op: Callable[[AsyncSession], Any], span_desc: str) -> Any async

Run op inside a managed session & sentry span (if enabled).

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
find_one(*, where: dict[str, Any], include: dict[str, bool] | None = None, **__: Any) -> ModelT | None async

Return the first row that matches where or None.

Source code in tux/database/controllers/starboard.py
Python
    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,
    },
find_many(*, where: dict[str, Any] | None = None, include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None) -> list[ModelT] async

Return a list of rows matching where (or all rows).

Source code in tux/database/controllers/starboard.py
Python
            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,
    and deleting starboard messages.
    """

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

    async def get_starboard_message(self, message_id: int, guild_id: int) -> StarboardMessage | None:
        """Get a starboard message by message ID and 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)
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}},
    )
execute_transaction(callback: Callable[[], Any]) -> Any async

Execute callback inside a database session / transaction block.

Source code in tux/database/controllers/starboard.py
Python
    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(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,
    )
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Return getattr(obj, attr, default) - keeps old helper available.

Source code in tux/database/controllers/starboard.py
Python
    The ID of the guild
limit : int | None
    Optional limit on the number of messages to return
order_by_stars : bool
connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any] staticmethod

Return a dict with a single key that can be merged into data dicts.

The calling code does something like::

Text Only
data = {"guild": connect_or_create_relation("guild_id", guild_id)}

We map that pattern to a very small helper that collapses to {"guild_id": guild_id}.

Source code in tux/database/controllers/starboard.py
Python
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,
)
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)
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,
    )
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})
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,
    )