reminder ¶

User reminder scheduling controller.

This controller manages scheduled reminders for Discord users, allowing them to set timed notifications and messages to be delivered at specified times.

Classes:

ReminderController –

Clean Reminder controller using the new BaseController pattern.

Classes¶

ReminderController ¶

Python

ReminderController(db: DatabaseService | None = None)

Bases: BaseController[Reminder]

Clean Reminder controller using the new BaseController pattern.

Initialize the reminder controller.

Parameters:

db (DatabaseService | None, default: None ) –

The database service instance. If None, uses the default service.

Returned by:

controllers Classes DatabaseCoordinator Attributes reminder

Methods:

get_reminder_by_id –

Get a reminder by its ID.
get_reminders_by_user –

Get all reminders for a specific user in a guild.
get_reminders_by_guild –

Get all reminders in a guild.
create_reminder –

Create a new reminder.
update_reminder –

Update a reminder by ID.
delete_reminder –

Delete a reminder by ID.
get_expired_reminders –

Get all expired reminders.
get_active_reminders –

Get all active (non-expired) reminders in a guild.
get_reminders_by_channel –

Get all reminders for a specific channel.
get_reminder_count_by_user –

Get the number of reminders for a user in a guild.
get_reminder_count_by_guild –

Get the total number of reminders in a guild.
create –

Create a new record.
delete_reminder_by_id –

Delete a reminder by its ID.
get_by_id –

Get a record by ID.
get_all_reminders –

Get all reminders in a guild.
update_by_id –

Update a record by ID.
insert_reminder –

Insert a new reminder - alias for create.
delete_by_id –

Delete a record by ID.
cleanup_expired_reminders –

Delete all expired reminders and return the count.
exists –

Check if a record exists.
find_one –

Find one record.
find_all –

Find all records with performance optimizations.
find_all_with_options –

Find all records with relationship loading options.
count –

Count records.
get_all –

Get all records (alias for find_all without pagination).
execute_query –

Execute a custom query.
find_with_json_query –

Find records using JSON column queries.
find_with_array_contains –

Find records where array column contains value.
find_with_full_text_search –

Find records using full-text search.
paginate –

Paginate records with metadata.
find_paginated –

Find paginated records with relationship loading.
bulk_create –

Create multiple records in bulk.
bulk_update –

Update multiple records in bulk.
bulk_delete –

Delete multiple records in bulk.
update_where –

Update records matching filters.
delete_where –

Delete records matching filters.
bulk_upsert_with_conflict_resolution –

Bulk upsert with conflict resolution.
with_session –

Execute operation within a session context.
with_transaction –

Execute operation within a transaction context.
execute_transaction –

Execute a callback within a transaction.
get_table_statistics –

Get comprehensive table statistics.
explain_query_performance –

Explain query performance with optional analysis.
upsert_by_field –

Upsert a record by a specific field.
upsert_by_id –

Upsert a record by ID.
get_or_create_by_field –

Get existing record or create new one by field.
get_or_create –

Get existing record or create new one.
upsert –

Upsert a record.

Attributes:

db_service (DatabaseService) –

Database service property for test compatibility.
model_class (type[ModelT]) –

Model class property for test compatibility.

Attributes¶

db_service `property` ¶

Python

db_service: DatabaseService

Database service property for test compatibility.

Returns:

DatabaseService –

The database service instance.

model_class `property` ¶

Python

model_class: type[ModelT]

Model class property for test compatibility.

Returns:

type[ModelT] –

The SQLModel class.

Functions¶

get_reminder_by_id `async` ¶

Python

get_reminder_by_id(reminder_id: int) -> Reminder | None

Get a reminder by its ID.

Returns:

Reminder | None –

The reminder if found, None otherwise.

get_reminders_by_user `async` ¶

Python

get_reminders_by_user(user_id: int, guild_id: int) -> list[Reminder]

Get all reminders for a specific user in a guild.

Returns:

list[Reminder] –

List of all reminders for the user in the guild.

get_reminders_by_guild `async` ¶

Python

get_reminders_by_guild(guild_id: int) -> list[Reminder]

Get all reminders in a guild.

Returns:

list[Reminder] –

List of all reminders for the guild.

create_reminder `async` ¶

Python

create_reminder(
    user_id: int,
    guild_id: int,
    channel_id: int,
    message: str,
    expires_at: datetime,
    **kwargs: Any,
) -> Reminder

Create a new reminder.

Returns:

Reminder –

The newly created reminder.

update_reminder `async` ¶

Python

update_reminder(reminder_id: int, **kwargs: Any) -> Reminder | None

Update a reminder by ID.

Returns:

Reminder | None –

The updated reminder, or None if not found.

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

delete_reminder `async` ¶

Python

delete_reminder(reminder_id: int) -> bool

Delete a reminder by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

get_expired_reminders `async` ¶

Python

get_expired_reminders() -> list[Reminder]

Get all expired reminders.

Returns:

list[Reminder] –

List of all expired reminders.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

get_active_reminders `async` ¶

Python

get_active_reminders(guild_id: int) -> list[Reminder]

Get all active (non-expired) reminders in a guild.

Returns:

list[Reminder] –

List of active reminders.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

get_reminders_by_channel `async` ¶

Python

get_reminders_by_channel(channel_id: int) -> list[Reminder]

Get all reminders for a specific channel.

Returns:

list[Reminder] –

List of reminders for the channel.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

get_reminder_count_by_user `async` ¶

Python

get_reminder_count_by_user(user_id: int, guild_id: int) -> int

Get the number of reminders for a user in a guild.

Returns:

int –

The count of reminders for the user.

get_reminder_count_by_guild `async` ¶

Python

get_reminder_count_by_guild(guild_id: int) -> int

Get the total number of reminders in a guild.

Returns:

int –

The total count of reminders in the guild.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

delete_reminder_by_id `async` ¶

Python

delete_reminder_by_id(reminder_id: int) -> bool

Delete a reminder by its ID.

Returns:

bool –

True if deleted successfully, False otherwise.

get_by_id `async` ¶

Python

get_by_id(record_id: Any) -> ModelT | None

Get a record by ID.

Returns:

ModelT | None –

The record if found, None otherwise.

get_all_reminders `async` ¶

Python

get_all_reminders(guild_id: int) -> list[Reminder]

Get all reminders in a guild.

Returns:

list[Reminder] –

List of all reminders for the guild.

update_by_id `async` ¶

Python

update_by_id(record_id: Any, **values: Any) -> ModelT | None

Update a record by ID.

Returns:

ModelT | None –

The updated record, or None if not found.

insert_reminder `async` ¶

Python

insert_reminder(**kwargs: Any) -> Reminder

Insert a new reminder - alias for create.

Returns:

Reminder –

The newly created reminder.

delete_by_id `async` ¶

Python

delete_by_id(record_id: Any) -> bool

Delete a record by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

cleanup_expired_reminders `async` ¶

Python

cleanup_expired_reminders() -> int

Delete all expired reminders and return the count.

Returns:

int –

The number of reminders that were deleted.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

find_one `async` ¶

Python

find_one(filters: Any | None = None, order_by: Any | None = None) -> ModelT | None

Find one record.

Returns:

ModelT | None –

The found record, or None if not found.

find_all `async` ¶

Python

find_all(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
) -> list[ModelT]

Find all records with performance optimizations.

Returns:

list[ModelT] –

List of found records.

find_all_with_options `async` ¶

Python

find_all_with_options(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
    load_relationships: list[str] | None = None,
) -> list[ModelT]

Find all records with relationship loading options.

Returns:

list[ModelT] –

List of found records with loaded relationships.

count `async` ¶

Python

count(filters: Any | None = None) -> int

Count records.

Returns:

int –

The count of matching records.

get_all `async` ¶

Python

get_all(filters: Any | None = None, order_by: Any | None = None) -> list[ModelT]

Get all records (alias for find_all without pagination).

Returns:

list[ModelT] –

List of all matching records.

execute_query `async` ¶

Python

execute_query(query: Any) -> Any

Execute a custom query.

Returns:

Any –

The query result.

find_with_json_query `async` ¶

Python

find_with_json_query(
    json_column: str, json_path: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records using JSON column queries.

Returns:

list[ModelT] –

List of records matching the JSON query.

find_with_array_contains `async` ¶

Python

find_with_array_contains(
    array_column: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records where array column contains value.

Returns:

list[ModelT] –

List of records with matching array values.

find_with_full_text_search `async` ¶

Python

find_with_full_text_search(
    search_columns: list[str], search_term: str, filters: Any | None = None
) -> list[ModelT]

Find records using full-text search.

Returns:

list[ModelT] –

List of records matching the search term.

paginate `async` ¶

Python

paginate(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
) -> PaginationResult[ModelT]

Paginate records with metadata.

Returns:

PaginationResult[ModelT] –

Pagination result with items, total, and page info.

find_paginated `async` ¶

Python

find_paginated(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
    load_relationships: list[str] | None = None,
) -> PaginationResult[ModelT]

Find paginated records with relationship loading.

Returns:

PaginationResult[ModelT] –

Pagination result with items and relationships loaded.

bulk_create `async` ¶

Python

bulk_create(items: list[dict[str, Any]]) -> list[ModelT]

Create multiple records in bulk.

Returns:

list[ModelT] –

List of created records.

bulk_update `async` ¶

Python

bulk_update(updates: list[tuple[Any, dict[str, Any]]]) -> int

Update multiple records in bulk.

Returns:

int –

Number of records updated.

bulk_delete `async` ¶

Python

bulk_delete(record_ids: list[Any]) -> int

Delete multiple records in bulk.

Returns:

int –

Number of records deleted.

update_where `async` ¶

Python

update_where(filters: Any, values: dict[str, Any]) -> int

Update records matching filters.

Returns:

int –

Number of records updated.

delete_where `async` ¶

Python

delete_where(filters: Any) -> int

Delete records matching filters.

Returns:

int –

Number of records deleted.

bulk_upsert_with_conflict_resolution `async` ¶

Python

bulk_upsert_with_conflict_resolution(
    items: list[dict[str, Any]],
    conflict_columns: list[str],
    update_columns: list[str] | None = None,
) -> list[ModelT]

Bulk upsert with conflict resolution.

Returns:

list[ModelT] –

List of upserted records.

with_session `async` ¶

Python

with_session[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a session context.

Returns:

R –

The result of the operation.

with_transaction `async` ¶

Python

with_transaction[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a transaction context.

Returns:

R –

The result of the operation.

execute_transaction `async` ¶

Python

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

Execute a callback within a transaction.

Returns:

Any –

The result of the callback.

get_table_statistics `async` ¶

Python

get_table_statistics() -> dict[str, Any]

Get comprehensive table statistics.

Returns:

dict[str, Any] –

Dictionary containing table statistics.

explain_query_performance `async` ¶

Python

explain_query_performance(
    query: Any, analyze: bool = False, buffers: bool = False
) -> dict[str, Any]

Explain query performance with optional analysis.

Returns:

dict[str, Any] –

Dictionary containing query execution plan and statistics.

upsert_by_field `async` ¶

Python

upsert_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Upsert a record by a specific field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert_by_id `async` ¶

Python

upsert_by_id(
    record_id: Any, defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record by ID.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create_by_field `async` ¶

Python

get_or_create_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Get existing record or create new one by field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create `async` ¶

Python

get_or_create(
    defaults: dict[str, Any] | None = None, **filters: Any
) -> tuple[ModelT, bool]

Get existing record or create new one.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert `async` ¶

Python

upsert(
    filters: dict[str, Any], defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

reminder ¶

Classes¶

ReminderController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

get_reminder_by_id async ¶

get_reminders_by_user async ¶

get_reminders_by_guild async ¶

create_reminder async ¶

update_reminder async ¶

_get_pagination ¶

delete_reminder async ¶

_get_bulk ¶

get_expired_reminders async ¶

_get_transaction ¶

get_active_reminders async ¶

_get_performance ¶

get_reminders_by_channel async ¶

_get_upsert ¶

get_reminder_count_by_user async ¶

get_reminder_count_by_guild async ¶

create async ¶

delete_reminder_by_id async ¶

get_by_id async ¶

get_all_reminders async ¶

update_by_id async ¶

insert_reminder async ¶

delete_by_id async ¶

cleanup_expired_reminders async ¶

exists async ¶

find_one async ¶

find_all async ¶

find_all_with_options async ¶

count async ¶

get_all async ¶

execute_query async ¶

find_with_json_query async ¶

find_with_array_contains async ¶

find_with_full_text_search async ¶

paginate async ¶

find_paginated async ¶

bulk_create async ¶

bulk_update async ¶

bulk_delete async ¶

update_where async ¶

delete_where async ¶

bulk_upsert_with_conflict_resolution async ¶

with_session async ¶

with_transaction async ¶

execute_transaction async ¶

get_table_statistics async ¶

explain_query_performance async ¶

upsert_by_field async ¶

upsert_by_id async ¶

get_or_create_by_field async ¶

get_or_create async ¶

upsert async ¶

db_service `property` ¶

model_class `property` ¶

get_reminder_by_id `async` ¶

get_reminders_by_user `async` ¶

get_reminders_by_guild `async` ¶

create_reminder `async` ¶

update_reminder `async` ¶

delete_reminder `async` ¶

get_expired_reminders `async` ¶

get_active_reminders `async` ¶

get_reminders_by_channel `async` ¶

get_reminder_count_by_user `async` ¶

get_reminder_count_by_guild `async` ¶

create `async` ¶

delete_reminder_by_id `async` ¶

get_by_id `async` ¶

get_all_reminders `async` ¶

update_by_id `async` ¶

insert_reminder `async` ¶

delete_by_id `async` ¶

cleanup_expired_reminders `async` ¶

exists `async` ¶

find_one `async` ¶

find_all `async` ¶

find_all_with_options `async` ¶

count `async` ¶

get_all `async` ¶

execute_query `async` ¶

find_with_json_query `async` ¶

find_with_array_contains `async` ¶

find_with_full_text_search `async` ¶

paginate `async` ¶

find_paginated `async` ¶

bulk_create `async` ¶

bulk_update `async` ¶

bulk_delete `async` ¶

update_where `async` ¶

delete_where `async` ¶

bulk_upsert_with_conflict_resolution `async` ¶

with_session `async` ¶

with_transaction `async` ¶

execute_transaction `async` ¶

get_table_statistics `async` ¶

explain_query_performance `async` ¶

upsert_by_field `async` ¶

upsert_by_id `async` ¶

get_or_create_by_field `async` ¶

get_or_create `async` ¶

upsert `async` ¶