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. |
insert_reminder | Create a new reminder. |
find_one | Return the first row that matches where or None. |
delete_reminder_by_id | Delete a reminder by its ID. |
find_many | Return a list of rows matching where (or all rows). |
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. |
count_reminders_by_guild_id | Count the number of reminders in a guild. |
execute_transaction | Execute callback inside a database session / transaction block. |
bulk_delete_reminders_by_guild_id | Delete all reminders for a guild. |
safe_get_attr | Return |
mark_reminders_as_sent | Mark multiple reminders as sent. |
connect_or_create_relation | Return a dict with a single key that can be merged into data dicts. |
Source code in tux/database/controllers/reminder.py
Functions¶
get_all_reminders() -> list[Reminder]
async
¶
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
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})
_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/reminder.py
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
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
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},
)
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/reminder.py
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},
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
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})
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/reminder.py
The deleted reminder if found, None otherwise
"""
return await self.delete(where={"reminder_id": reminder_id})
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},
)
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
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},
)
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
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},
)
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
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)
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
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)
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
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)
execute_transaction(callback: Callable[[], Any]) -> Any
async
¶
Execute callback inside a database session / transaction block.
Source code in tux/database/controllers/reminder.py
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
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})
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any
staticmethod
¶
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
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})
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::
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}
.