Skip to content

tux.database.controllers.reminder

Classes:

Name Description
ReminderController

Controller for managing user reminders.

Classes

ReminderController()

Bases: BaseController[Reminder]

Controller for managing user reminders.

This controller provides methods for creating, retrieving, updating, and deleting reminders for users across guilds.

Initialize the ReminderController with the reminder table.

Methods:

Name Description
get_all_reminders

Get all reminders across all guilds.
get_reminder_by_id

Get a reminder by its ID.
get_unsent_reminders

Get all unsent reminders that have expired.
insert_reminder

Create a new reminder.
delete_reminder_by_id

Delete a reminder by its ID.
update_reminder_by_id

Update a reminder's content.
update_reminder_status

Update the status of a reminder.
get_reminders_by_user_id

Get all reminders for a user.
get_reminders_by_guild_id

Get all reminders for a guild.
find_one

Finds the first record matching specified criteria.
count_reminders_by_guild_id

Count the number of reminders in a guild.
find_unique

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

Delete all reminders for a guild.
mark_reminders_as_sent

Mark multiple reminders as sent.
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/reminder.py
Python
def __init__(self) -> None:
    """Initialize the ReminderController with the reminder table."""
    super().__init__("reminder")
    self.guild_table: GuildActions[Guild] = db.client.guild

Functions

get_all_reminders() -> list[Reminder] async

Get all reminders across all guilds.

Returns:

Type Description
list[Reminder]

List of all reminders
Source code in tux/database/controllers/reminder.py
Python
async def get_all_reminders(self) -> list[Reminder]:
    """Get all reminders across all guilds.

    Returns
    -------
    list[Reminder]
        List of all reminders
    """
    return await self.find_many(where={})
get_reminder_by_id(reminder_id: int) -> Reminder | None async

Get a reminder by its ID.

Parameters:

Name Type Description Default
reminder_id int

The ID of the reminder to get

 required

Returns:

Type Description
Reminder | None

The reminder if found, None otherwise
Source code in tux/database/controllers/reminder.py
Python
async def get_reminder_by_id(self, reminder_id: int) -> Reminder | None:
    """Get a reminder by its ID.

    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to get

    Returns
    -------
    Reminder | None
        The reminder if found, None otherwise
    """
    return await self.find_unique(where={"reminder_id": reminder_id})
get_unsent_reminders() -> list[Reminder] async

Get all unsent reminders that have expired.

This method finds reminders that should be sent to users because their expiration time has passed.

Returns:

Type Description
list[Reminder]

List of unsent expired reminders
Source code in tux/database/controllers/reminder.py
Python
async def get_unsent_reminders(self) -> list[Reminder]:
    """Get all unsent reminders that have expired.

    This method finds reminders that should be sent to users
    because their expiration time has passed.

    Returns
    -------
    list[Reminder]
        List of unsent expired reminders
    """
    now = datetime.now(UTC)
    return await self.find_many(where={"reminder_sent": False, "reminder_expires_at": {"lte": now}})
insert_reminder(reminder_user_id: int, reminder_content: str, reminder_expires_at: datetime, reminder_channel_id: int, guild_id: int) -> Reminder async

Create a new reminder.

Parameters:

Name Type Description Default
reminder_user_id int

The ID of the user to remind

 required
reminder_content str

The content of the reminder

 required
reminder_expires_at datetime

When the reminder should be sent

 required
reminder_channel_id int

The ID of the channel to send the reminder to

 required
guild_id int

The ID of the guild the reminder belongs to

 required

Returns:

Type Description
Reminder

The created reminder
Source code in tux/database/controllers/reminder.py
Python
async def insert_reminder(
    self,
    reminder_user_id: int,
    reminder_content: str,
    reminder_expires_at: datetime,
    reminder_channel_id: int,
    guild_id: int,
) -> Reminder:
    """Create a new reminder.

    Parameters
    ----------
    reminder_user_id : int
        The ID of the user to remind
    reminder_content : str
        The content of the reminder
    reminder_expires_at : datetime
        When the reminder should be sent
    reminder_channel_id : int
        The ID of the channel to send the reminder to
    guild_id : int
        The ID of the guild the reminder belongs to

    Returns
    -------
    Reminder
        The created reminder
    """
    return await self.create(
        data={
            "reminder_user_id": reminder_user_id,
            "reminder_content": reminder_content,
            "reminder_expires_at": reminder_expires_at,
            "reminder_channel_id": reminder_channel_id,
            "reminder_sent": False,
            "guild": self.connect_or_create_relation("guild_id", guild_id),
        },
        include={"guild": True},
    )
_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/reminder.py
Python
    Parameters
    ----------
    reminder_user_id : int
        The ID of the user to remind
    reminder_content : str
        The content of the reminder
    reminder_expires_at : datetime
        When the reminder should be sent
    reminder_channel_id : int
        The ID of the channel to send the reminder to
    guild_id : int
        The ID of the guild the reminder belongs to

    Returns
    -------
    Reminder
        The created reminder
    """
    return await self.create(
        data={
            "reminder_user_id": reminder_user_id,
            "reminder_content": reminder_content,
            "reminder_expires_at": reminder_expires_at,
            "reminder_channel_id": reminder_channel_id,
            "reminder_sent": False,
            "guild": self.connect_or_create_relation("guild_id", guild_id),
        },
        include={"guild": True},
    )

async def delete_reminder_by_id(self, reminder_id: int) -> Reminder | None:
    """Delete a reminder by its ID.

    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to delete

    Returns
    -------
    Reminder | None
        The deleted reminder if found, None otherwise
    """
    return await self.delete(where={"reminder_id": reminder_id})

async def update_reminder_by_id(
delete_reminder_by_id(reminder_id: int) -> Reminder | None async

Delete a reminder by its ID.

Parameters:

Name Type Description Default
reminder_id int

The ID of the reminder to delete

 required

Returns:

Type Description
Reminder | None

The deleted reminder if found, None otherwise
Source code in tux/database/controllers/reminder.py
Python
async def delete_reminder_by_id(self, reminder_id: int) -> Reminder | None:
    """Delete a reminder by its ID.

    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to delete

    Returns
    -------
    Reminder | None
        The deleted reminder if found, None otherwise
    """
    return await self.delete(where={"reminder_id": reminder_id})
update_reminder_by_id(reminder_id: int, reminder_content: str) -> Reminder | None async

Update a reminder's content.

Parameters:

Name Type Description Default
reminder_id int

The ID of the reminder to update

 required
reminder_content str

The new content for the reminder

 required

Returns:

Type Description
Reminder | None

The updated reminder if found, None otherwise
Source code in tux/database/controllers/reminder.py
Python
async def update_reminder_by_id(
    self,
    reminder_id: int,
    reminder_content: str,
) -> Reminder | None:
    """Update a reminder's content.

    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to update
    reminder_content : str
        The new content for the reminder

    Returns
    -------
    Reminder | None
        The updated reminder if found, None otherwise
    """
    return await self.update(
        where={"reminder_id": reminder_id},
        data={"reminder_content": reminder_content},
    )
_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/reminder.py
Python
    reminder_id: int,
    reminder_content: str,
) -> Reminder | None:
    """Update a reminder's content.
_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/reminder.py
Python
    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to update
    reminder_content : str
        The new content for the reminder

    Returns
    -------
    Reminder | None
        The updated reminder if found, None otherwise
    """
    return await self.update(
        where={"reminder_id": reminder_id},
        data={"reminder_content": reminder_content},
    )

async def update_reminder_status(self, reminder_id: int, sent: bool = True) -> Reminder | None:
    """Update the status of a reminder.

    This method sets the value "reminder_sent" to True by default.
update_reminder_status(reminder_id: int, sent: bool = True) -> Reminder | None async

Update the status of a reminder.

This method sets the value "reminder_sent" to True by default.

Parameters:

Name Type Description Default
reminder_id int

The ID of the reminder to update

 required
sent bool

The new status of the reminder

 True

Returns:

Type Description
Reminder | None

The updated reminder if found, None otherwise
Source code in tux/database/controllers/reminder.py
Python
async def update_reminder_status(self, reminder_id: int, sent: bool = True) -> Reminder | None:
    """Update the status of a reminder.

    This method sets the value "reminder_sent" to True by default.

    Parameters
    ----------
    reminder_id : int
        The ID of the reminder to update
    sent : bool
        The new status of the reminder

    Returns
    -------
    Reminder | None
        The updated reminder if found, None otherwise
    """
    return await self.update(
        where={"reminder_id": reminder_id},
        data={"reminder_sent": sent},
    )
_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/reminder.py
Python
Parameters
----------
reminder_id : int
    The ID of the reminder to update
sent : bool
    The new status of the reminder

Returns
-------
Reminder | None
_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/reminder.py
Python
    """
    return await self.update(
        where={"reminder_id": reminder_id},
        data={"reminder_sent": sent},
    )

async def get_reminders_by_user_id(
get_reminders_by_user_id(user_id: int, include_sent: bool = False, limit: int | None = None) -> list[Reminder] async

Get all reminders for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to get reminders for

 required
include_sent bool

Whether to include reminders that have already been sent

 False
limit int | None

Optional limit on the number of reminders to return

 None

Returns:

Type Description
list[Reminder]

List of reminders for the user
Source code in tux/database/controllers/reminder.py
Python
async def get_reminders_by_user_id(
    self,
    user_id: int,
    include_sent: bool = False,
    limit: int | None = None,
) -> list[Reminder]:
    """Get all reminders for a user.

    Parameters
    ----------
    user_id : int
        The ID of the user to get reminders for
    include_sent : bool
        Whether to include reminders that have already been sent
    limit : int | None
        Optional limit on the number of reminders to return

    Returns
    -------
    list[Reminder]
        List of reminders for the user
    """
    where = {"reminder_user_id": user_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.find_many(where=where, order={"reminder_expires_at": "asc"}, take=limit)
_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/reminder.py
Python
    user_id: int,
    include_sent: bool = False,
    limit: int | None = None,
) -> list[Reminder]:
    """Get all reminders for a user.

    Parameters
    ----------
    user_id : int
        The ID of the user to get reminders for
_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/reminder.py
Python
    Whether to include reminders that have already been sent
limit : int | None
    Optional limit on the number of reminders to return

Returns
-------
list[Reminder]
_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/reminder.py
Python
    """
    where = {"reminder_user_id": user_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.find_many(where=where, order={"reminder_expires_at": "asc"}, take=limit)

async def get_reminders_by_guild_id(
    self,
    guild_id: int,
    include_sent: bool = False,
    limit: int | None = None,
) -> list[Reminder]:
    """Get all reminders for a guild.

    Parameters
    ----------
get_reminders_by_guild_id(guild_id: int, include_sent: bool = False, limit: int | None = None) -> list[Reminder] async

Get all reminders for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to get reminders for

 required
include_sent bool

Whether to include reminders that have already been sent

 False
limit int | None

Optional limit on the number of reminders to return

 None

Returns:

Type Description
list[Reminder]

List of reminders for the guild
Source code in tux/database/controllers/reminder.py
Python
async def get_reminders_by_guild_id(
    self,
    guild_id: int,
    include_sent: bool = False,
    limit: int | None = None,
) -> list[Reminder]:
    """Get all reminders for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get reminders for
    include_sent : bool
        Whether to include reminders that have already been sent
    limit : int | None
        Optional limit on the number of reminders to return

    Returns
    -------
    list[Reminder]
        List of reminders for the guild
    """
    where = {"guild_id": guild_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.find_many(where=where, order={"reminder_expires_at": "asc"}, take=limit)
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Reminder | 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/reminder.py
Python
        Whether to include reminders that have already been sent
    limit : int | None
        Optional limit on the number of reminders to return

    Returns
    -------
    list[Reminder]
        List of reminders for the guild
    """
    where = {"guild_id": guild_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.find_many(where=where, order={"reminder_expires_at": "asc"}, take=limit)

async def count_reminders_by_guild_id(self, guild_id: int, include_sent: bool = False) -> int:
    """Count the number of reminders in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to count reminders for
    include_sent : bool
        Whether to include reminders that have already been sent

    Returns
    -------
count_reminders_by_guild_id(guild_id: int, include_sent: bool = False) -> int async

Count the number of reminders in a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to count reminders for

 required
include_sent bool

Whether to include reminders that have already been sent

 False

Returns:

Type Description
int

The number of reminders in the guild
Source code in tux/database/controllers/reminder.py
Python
async def count_reminders_by_guild_id(self, guild_id: int, include_sent: bool = False) -> int:
    """Count the number of reminders in a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to count reminders for
    include_sent : bool
        Whether to include reminders that have already been sent

    Returns
    -------
    int
        The number of reminders in the guild
    """
    where = {"guild_id": guild_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.count(where=where)
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Reminder | 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/reminder.py
Python
        The number of reminders in the guild
    """
    where = {"guild_id": guild_id}
    if not include_sent:
        where["reminder_sent"] = False

    return await self.count(where=where)

async def bulk_delete_reminders_by_guild_id(self, guild_id: int) -> int:
    """Delete all reminders for a guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete reminders for

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

async def mark_reminders_as_sent(self, reminder_ids: list[int]) -> int:
bulk_delete_reminders_by_guild_id(guild_id: int) -> int async

Delete all reminders for a guild.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to delete reminders for

 required

Returns:

Type Description
int

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

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete reminders for

    Returns
    -------
    int
        The number of reminders deleted
    """
    return await self.delete_many(where={"guild_id": guild_id})
mark_reminders_as_sent(reminder_ids: list[int]) -> int async

Mark multiple reminders as sent.

Parameters:

Name Type Description Default
reminder_ids list[int]

The IDs of the reminders to mark as sent

 required

Returns:

Type Description
int

The number of reminders updated
Source code in tux/database/controllers/reminder.py
Python
async def mark_reminders_as_sent(self, reminder_ids: list[int]) -> int:
    """Mark multiple reminders as sent.

    Parameters
    ----------
    reminder_ids : list[int]
        The IDs of the reminders to mark as sent

    Returns
    -------
    int
        The number of reminders updated
    """
    return await self.update_many(where={"reminder_id": {"in": reminder_ids}}, data={"reminder_sent": True})
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[Reminder] 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/reminder.py
Python
Parameters
----------
reminder_ids : list[int]
    The IDs of the reminders to mark as sent

Returns
-------
int
    The number of reminders updated
"""
return await self.update_many(where={"reminder_id": {"in": reminder_ids}}, data={"reminder_sent": True})
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.
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Reminder 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.
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Reminder | 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.
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Reminder | 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.
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Reminder 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.
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.