tux.database.controllers.snippet
¶
Classes:
Name | Description |
---|---|
SnippetController | Controller for managing snippets. |
Classes¶
SnippetController()
¶
Bases: BaseController[Snippet]
Controller for managing snippets.
This controller provides methods for managing snippet records in the database. It inherits common CRUD operations from BaseController.
Initialize the SnippetController with the snippet table.
Methods:
Name | Description |
---|---|
get_all_snippets | Get all snippets. |
get_all_snippets_by_guild_id | Get all snippets for a guild. |
get_all_snippets_sorted | Get all snippets sorted by creation time. |
get_snippet_by_name | Get a snippet by name. |
get_snippet_by_name_and_guild_id | Get a snippet by name and guild ID. |
create_snippet | Create a new snippet. |
get_snippet_by_id | Get a snippet by its ID. |
delete_snippet_by_id | Delete a snippet by its ID. |
create_snippet_alias | Create a new snippet alias. |
find_one | Finds the first record matching specified criteria. |
find_unique | Finds a single record by a unique constraint (e.g., ID). |
get_all_aliases | Get all aliases for a snippet name within a guild. |
update_snippet_by_id | Update a snippet's content. |
find_many | Finds multiple records matching specified criteria. |
increment_snippet_uses | Increment the use counter for a snippet. |
count | Counts records matching the specified criteria. |
lock_snippet_by_id | Lock a snippet. |
create | Creates a new record in the table. |
unlock_snippet_by_id | Unlock a snippet. |
toggle_snippet_lock_by_id | Toggle a snippet's lock state. |
update | Updates a single existing record matching the criteria. |
count_snippets_by_guild_id | Count the number of snippets in a guild. |
delete | Deletes a single record matching the criteria. |
bulk_delete_snippets_by_guild_id | Delete all snippets for a guild. |
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/snippet.py
Functions¶
get_all_snippets() -> list[Snippet]
async
¶
get_all_snippets_by_guild_id(guild_id: int, include_guild: bool = False) -> list[Snippet]
async
¶
Get all snippets for a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id | int | The ID of the guild to get snippets for | required |
include_guild | bool | Whether to include the guild relation | False |
Returns:
Type | Description |
---|---|
list[Snippet] | List of snippets for the guild |
Source code in tux/database/controllers/snippet.py
async def get_all_snippets_by_guild_id(self, guild_id: int, include_guild: bool = False) -> list[Snippet]:
"""Get all snippets for a guild.
Parameters
----------
guild_id : int
The ID of the guild to get snippets for
include_guild : bool
Whether to include the guild relation
Returns
-------
list[Snippet]
List of snippets for the guild
"""
include = {"guild": True} if include_guild else None
return await self.find_many(where={"guild_id": guild_id}, include=include)
get_all_snippets_sorted(newestfirst: bool = True, limit: int | None = None) -> list[Snippet]
async
¶
Get all snippets sorted by creation time.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
newestfirst | bool | Whether to sort with newest first | True |
limit | int | None | Optional maximum number of snippets to return | None |
Returns:
Type | Description |
---|---|
list[Snippet] | List of sorted snippets |
Source code in tux/database/controllers/snippet.py
async def get_all_snippets_sorted(self, newestfirst: bool = True, limit: int | None = None) -> list[Snippet]:
"""Get all snippets sorted by creation time.
Parameters
----------
newestfirst : bool
Whether to sort with newest first
limit : int | None
Optional maximum number of snippets to return
Returns
-------
list[Snippet]
List of sorted snippets
"""
return await self.find_many(
where={},
order={"snippet_created_at": "desc" if newestfirst else "asc"},
take=limit,
)
get_snippet_by_name(snippet_name: str, include_guild: bool = False) -> Snippet | None
async
¶
Get a snippet by name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_name | str | The name of the snippet to get | required |
include_guild | bool | Whether to include the guild relation | False |
Returns:
Type | Description |
---|---|
Snippet | None | The snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def get_snippet_by_name(self, snippet_name: str, include_guild: bool = False) -> Snippet | None:
"""Get a snippet by name.
Parameters
----------
snippet_name : str
The name of the snippet to get
include_guild : bool
Whether to include the guild relation
Returns
-------
Snippet | None
The snippet if found, None otherwise
"""
include = {"guild": True} if include_guild else None
return await self.find_one(
where={"snippet_name": {"contains": snippet_name, "mode": "insensitive"}},
include=include,
)
_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/snippet.py
async def get_snippet_by_name(self, snippet_name: str, include_guild: bool = False) -> Snippet | None:
"""Get a snippet by name.
Parameters
----------
snippet_name : str
The name of the snippet to get
include_guild : bool
Whether to include the guild relation
Returns
-------
Snippet | None
The snippet if found, None otherwise
"""
include = {"guild": True} if include_guild else None
return await self.find_one(
where={"snippet_name": {"contains": snippet_name, "mode": "insensitive"}},
include=include,
)
async def get_snippet_by_name_and_guild_id(
self,
snippet_name: str,
guild_id: int,
include_guild: bool = False,
) -> Snippet | None:
"""Get a snippet by name and guild ID.
Parameters
----------
snippet_name : str
The name of the snippet to get
guild_id : int
The ID of the guild to get the snippet from
include_guild : bool
Whether to include the guild relation
Returns
-------
Snippet | None
The snippet if found, None otherwise
"""
include = {"guild": True} if include_guild else None
return await self.find_one(
where={"snippet_name": {"equals": snippet_name, "mode": "insensitive"}, "guild_id": guild_id},
get_snippet_by_name_and_guild_id(snippet_name: str, guild_id: int, include_guild: bool = False) -> Snippet | None
async
¶
Get a snippet by name and guild ID.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_name | str | The name of the snippet to get | required |
guild_id | int | The ID of the guild to get the snippet from | required |
include_guild | bool | Whether to include the guild relation | False |
Returns:
Type | Description |
---|---|
Snippet | None | The snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def get_snippet_by_name_and_guild_id(
self,
snippet_name: str,
guild_id: int,
include_guild: bool = False,
) -> Snippet | None:
"""Get a snippet by name and guild ID.
Parameters
----------
snippet_name : str
The name of the snippet to get
guild_id : int
The ID of the guild to get the snippet from
include_guild : bool
Whether to include the guild relation
Returns
-------
Snippet | None
The snippet if found, None otherwise
"""
include = {"guild": True} if include_guild else None
return await self.find_one(
where={"snippet_name": {"equals": snippet_name, "mode": "insensitive"}, "guild_id": guild_id},
include=include,
)
_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None
¶
create_snippet(snippet_name: str, snippet_content: str, snippet_created_at: datetime.datetime, snippet_user_id: int, guild_id: int) -> Snippet
async
¶
Create a new snippet.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_name | str | The name of the snippet | required |
snippet_content | str | The content of the snippet | required |
snippet_created_at | datetime | The creation time of the snippet | required |
snippet_user_id | int | The ID of the user creating the snippet | required |
guild_id | int | The ID of the guild the snippet belongs to | required |
Returns:
Type | Description |
---|---|
Snippet | The created snippet |
Source code in tux/database/controllers/snippet.py
async def create_snippet(
self,
snippet_name: str,
snippet_content: str,
snippet_created_at: datetime.datetime,
snippet_user_id: int,
guild_id: int,
) -> Snippet:
"""Create a new snippet.
Parameters
----------
snippet_name : str
The name of the snippet
snippet_content : str
The content of the snippet
snippet_created_at : datetime.datetime
The creation time of the snippet
snippet_user_id : int
The ID of the user creating the snippet
guild_id : int
The ID of the guild the snippet belongs to
Returns
-------
Snippet
The created snippet
"""
# Use connect_or_create pattern instead of ensure_guild_exists
return await self.create(
data={
"snippet_name": snippet_name,
"snippet_content": snippet_content,
"snippet_created_at": snippet_created_at,
"snippet_user_id": snippet_user_id,
"guild": self.connect_or_create_relation("guild_id", guild_id),
"uses": 0,
"locked": False,
},
include={"guild": True},
)
_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/snippet.py
snippet_content: str,
snippet_created_at: datetime.datetime,
snippet_user_id: int,
guild_id: int,
) -> Snippet:
"""Create a new snippet.
Parameters
----------
snippet_name : str
The name of the snippet
snippet_content : str
The content of the snippet
snippet_created_at : datetime.datetime
The creation time of the snippet
snippet_user_id : int
The ID of the user creating the snippet
guild_id : int
The ID of the guild the snippet belongs to
Returns
_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/snippet.py
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]
¶
get_snippet_by_id(snippet_id: int, include_guild: bool = False) -> Snippet | None
async
¶
Get a snippet by its ID.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to get | required |
include_guild | bool | Whether to include the guild relation | False |
Returns:
Type | Description |
---|---|
Snippet | None | The snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def get_snippet_by_id(self, snippet_id: int, include_guild: bool = False) -> Snippet | None:
"""Get a snippet by its ID.
Parameters
----------
snippet_id : int
The ID of the snippet to get
include_guild : bool
Whether to include the guild relation
Returns
-------
Snippet | None
The snippet if found, None otherwise
"""
include = {"guild": True} if include_guild else None
return await self.find_unique(where={"snippet_id": snippet_id}, include=include)
_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]
¶
Constructs keyword arguments for Prisma delete operations.
Source code in tux/database/controllers/snippet.py
delete_snippet_by_id(snippet_id: int) -> Snippet | None
async
¶
Delete a snippet by its ID.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to delete | required |
Returns:
Type | Description |
---|---|
Snippet | None | The deleted snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def delete_snippet_by_id(self, snippet_id: int) -> Snippet | None:
"""Delete a snippet by its ID.
Parameters
----------
snippet_id : int
The ID of the snippet to delete
Returns
-------
Snippet | None
The deleted snippet if found, None otherwise
"""
return await self.delete(where={"snippet_id": snippet_id})
_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/snippet.py
Parameters
----------
snippet_id : int
The ID of the snippet to delete
Returns
-------
Snippet | None
The deleted snippet if found, None otherwise
"""
return await self.delete(where={"snippet_id": snippet_id})
async def create_snippet_alias(
self,
snippet_name: str,
snippet_alias: str,
snippet_created_at: datetime.datetime,
create_snippet_alias(snippet_name: str, snippet_alias: str, snippet_created_at: datetime.datetime, snippet_user_id: int, guild_id: int) -> Snippet
async
¶
Create a new snippet alias.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_name | str | The name of the snippet this is an alias for. | required |
snippet_alias | str | The alias name. | required |
snippet_created_at | datetime | The creation time of the alias. | required |
snippet_user_id | int | The ID of the user creating the alias. | required |
guild_id | int | The ID of the guild the alias belongs to. | required |
Returns:
Type | Description |
---|---|
Snippet | The created snippet alias record. |
Source code in tux/database/controllers/snippet.py
async def create_snippet_alias(
self,
snippet_name: str,
snippet_alias: str,
snippet_created_at: datetime.datetime,
snippet_user_id: int,
guild_id: int,
) -> Snippet:
"""Create a new snippet alias.
Parameters
----------
snippet_name : str
The name of the snippet this is an alias for.
snippet_alias : str
The alias name.
snippet_created_at : datetime.datetime
The creation time of the alias.
snippet_user_id : int
The ID of the user creating the alias.
guild_id : int
The ID of the guild the alias belongs to.
Returns
-------
Snippet
The created snippet alias record.
"""
# Use connect_or_create pattern for guild relation
return await self.create(
data={
"snippet_name": snippet_name,
"alias": snippet_alias, # Assuming 'alias' is the correct field name
"snippet_created_at": snippet_created_at,
"snippet_user_id": snippet_user_id,
"guild": self.connect_or_create_relation("guild_id", guild_id),
"uses": 0, # Set default values
"locked": False,
},
include={"guild": True},
)
find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Snippet | 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/snippet.py
"""Create a new snippet alias.
Parameters
----------
snippet_name : str
The name of the snippet this is an alias for.
snippet_alias : str
The alias name.
snippet_created_at : datetime.datetime
The creation time of the alias.
snippet_user_id : int
The ID of the user creating the alias.
guild_id : int
The ID of the guild the alias belongs to.
Returns
-------
Snippet
The created snippet alias record.
"""
# Use connect_or_create pattern for guild relation
return await self.create(
data={
"snippet_name": snippet_name,
"alias": snippet_alias, # Assuming 'alias' is the correct field name
"snippet_created_at": snippet_created_at,
"snippet_user_id": snippet_user_id,
find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Snippet | 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/snippet.py
"uses": 0, # Set default values
"locked": False,
},
include={"guild": True},
)
async def get_all_aliases(self, snippet_name: str, guild_id: int) -> list[Snippet]:
"""Get all aliases for a snippet name within a guild.
Parameters
----------
snippet_name : str
The name of the snippet to find aliases for.
guild_id : int
The ID of the guild to search within.
Returns
-------
list[Snippet]
A list of Snippet objects representing the aliases.
"""
return await self.find_many(
where={"alias": {"equals": snippet_name, "mode": "insensitive"}, "guild_id": guild_id},
)
get_all_aliases(snippet_name: str, guild_id: int) -> list[Snippet]
async
¶
Get all aliases for a snippet name within a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_name | str | The name of the snippet to find aliases for. | required |
guild_id | int | The ID of the guild to search within. | required |
Returns:
Type | Description |
---|---|
list[Snippet] | A list of Snippet objects representing the aliases. |
Source code in tux/database/controllers/snippet.py
async def get_all_aliases(self, snippet_name: str, guild_id: int) -> list[Snippet]:
"""Get all aliases for a snippet name within a guild.
Parameters
----------
snippet_name : str
The name of the snippet to find aliases for.
guild_id : int
The ID of the guild to search within.
Returns
-------
list[Snippet]
A list of Snippet objects representing the aliases.
"""
return await self.find_many(
where={"alias": {"equals": snippet_name, "mode": "insensitive"}, "guild_id": guild_id},
)
update_snippet_by_id(snippet_id: int, snippet_content: str) -> Snippet | None
async
¶
Update a snippet's content.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to update | required |
snippet_content | str | The new content for the snippet | required |
Returns:
Type | Description |
---|---|
Snippet | None | The updated snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def update_snippet_by_id(self, snippet_id: int, snippet_content: str) -> Snippet | None:
"""Update a snippet's content.
Parameters
----------
snippet_id : int
The ID of the snippet to update
snippet_content : str
The new content for the snippet
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
return await self.update(
where={"snippet_id": snippet_id},
data={"snippet_content": snippet_content},
)
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[Snippet]
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/snippet.py
async def update_snippet_by_id(self, snippet_id: int, snippet_content: str) -> Snippet | None:
"""Update a snippet's content.
Parameters
----------
snippet_id : int
The ID of the snippet to update
snippet_content : str
The new content for the snippet
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
return await self.update(
where={"snippet_id": snippet_id},
data={"snippet_content": snippet_content},
)
async def increment_snippet_uses(self, snippet_id: int) -> Snippet | None:
"""Increment the use counter for a snippet.
This method uses a transaction to ensure atomicity.
Parameters
----------
snippet_id : int
The ID of the snippet to increment
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
async def increment_tx():
snippet = await self.find_unique(where={"snippet_id": snippet_id})
if snippet is None:
return None
# Safely get the current uses value
snippet_uses = self.safe_get_attr(snippet, "uses", 0)
increment_snippet_uses(snippet_id: int) -> Snippet | None
async
¶
Increment the use counter for a snippet.
This method uses a transaction to ensure atomicity.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to increment | required |
Returns:
Type | Description |
---|---|
Snippet | None | The updated snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def increment_snippet_uses(self, snippet_id: int) -> Snippet | None:
"""Increment the use counter for a snippet.
This method uses a transaction to ensure atomicity.
Parameters
----------
snippet_id : int
The ID of the snippet to increment
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
async def increment_tx():
snippet = await self.find_unique(where={"snippet_id": snippet_id})
if snippet is None:
return None
# Safely get the current uses value
snippet_uses = self.safe_get_attr(snippet, "uses", 0)
return await self.update(
where={"snippet_id": snippet_id},
data={"uses": snippet_uses + 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/snippet.py
return await self.update(
where={"snippet_id": snippet_id},
data={"uses": snippet_uses + 1},
)
return await self.execute_transaction(increment_tx)
async def lock_snippet_by_id(self, snippet_id: int) -> Snippet | None:
"""Lock a snippet.
Parameters
----------
snippet_id : int
The ID of the snippet to lock
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
lock_snippet_by_id(snippet_id: int) -> Snippet | None
async
¶
Lock a snippet.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to lock | required |
Returns:
Type | Description |
---|---|
Snippet | None | The updated snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def lock_snippet_by_id(self, snippet_id: int) -> Snippet | None:
"""Lock a snippet.
Parameters
----------
snippet_id : int
The ID of the snippet to lock
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
return await self.update(
where={"snippet_id": snippet_id},
data={"locked": True},
)
create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Snippet
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/snippet.py
where={"snippet_id": snippet_id},
data={"locked": True},
)
async def unlock_snippet_by_id(self, snippet_id: int) -> Snippet | None:
"""Unlock a snippet.
Parameters
----------
snippet_id : int
The ID of the snippet to unlock
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
return await self.update(
where={"snippet_id": snippet_id},
data={"locked": False},
)
async def toggle_snippet_lock_by_id(self, snippet_id: int) -> Snippet | None:
"""Toggle a snippet's lock state.
unlock_snippet_by_id(snippet_id: int) -> Snippet | None
async
¶
Unlock a snippet.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to unlock | required |
Returns:
Type | Description |
---|---|
Snippet | None | The updated snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def unlock_snippet_by_id(self, snippet_id: int) -> Snippet | None:
"""Unlock a snippet.
Parameters
----------
snippet_id : int
The ID of the snippet to unlock
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
return await self.update(
where={"snippet_id": snippet_id},
data={"locked": False},
)
toggle_snippet_lock_by_id(snippet_id: int) -> Snippet | None
async
¶
Toggle a snippet's lock state.
This method uses a transaction to ensure atomicity.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
snippet_id | int | The ID of the snippet to toggle | required |
Returns:
Type | Description |
---|---|
Snippet | None | The updated snippet if found, None otherwise |
Source code in tux/database/controllers/snippet.py
async def toggle_snippet_lock_by_id(self, snippet_id: int) -> Snippet | None:
"""Toggle a snippet's lock state.
This method uses a transaction to ensure atomicity.
Parameters
----------
snippet_id : int
The ID of the snippet to toggle
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
async def toggle_lock_tx():
snippet = await self.find_unique(where={"snippet_id": snippet_id})
if snippet is None:
return None
# Safely get the current locked state
is_locked = self.safe_get_attr(snippet, "locked", False)
return await self.update(
where={"snippet_id": snippet_id},
data={"locked": not is_locked},
)
return await self.execute_transaction(toggle_lock_tx)
update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Snippet | 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/snippet.py
This method uses a transaction to ensure atomicity.
Parameters
----------
snippet_id : int
The ID of the snippet to toggle
Returns
-------
Snippet | None
The updated snippet if found, None otherwise
"""
async def toggle_lock_tx():
snippet = await self.find_unique(where={"snippet_id": snippet_id})
if snippet is None:
return None
# Safely get the current locked state
is_locked = self.safe_get_attr(snippet, "locked", False)
return await self.update(
where={"snippet_id": snippet_id},
data={"locked": not is_locked},
)
return await self.execute_transaction(toggle_lock_tx)
count_snippets_by_guild_id(guild_id: int) -> int
async
¶
Count the number of snippets in a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id | int | The ID of the guild to count snippets for | required |
Returns:
Type | Description |
---|---|
int | The number of snippets in the guild |
Source code in tux/database/controllers/snippet.py
async def count_snippets_by_guild_id(self, guild_id: int) -> int:
"""Count the number of snippets in a guild.
Parameters
----------
guild_id : int
The ID of the guild to count snippets for
Returns
-------
int
The number of snippets in the guild
"""
return await self.count(where={"guild_id": guild_id})
delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Snippet | 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/snippet.py
async def count_snippets_by_guild_id(self, guild_id: int) -> int:
"""Count the number of snippets in a guild.
Parameters
----------
guild_id : int
The ID of the guild to count snippets for
Returns
-------
int
The number of snippets in the guild
"""
return await self.count(where={"guild_id": guild_id})
async def bulk_delete_snippets_by_guild_id(self, guild_id: int) -> int:
"""Delete all snippets for a guild.
Parameters
----------
guild_id : int
The ID of the guild to delete snippets for
Returns
bulk_delete_snippets_by_guild_id(guild_id: int) -> int
async
¶
Delete all snippets for a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id | int | The ID of the guild to delete snippets for | required |
Returns:
Type | Description |
---|---|
int | The number of snippets deleted |
Source code in tux/database/controllers/snippet.py
async def bulk_delete_snippets_by_guild_id(self, guild_id: int) -> int:
"""Delete all snippets for a guild.
Parameters
----------
guild_id : int
The ID of the guild to delete snippets for
Returns
-------
int
The number of snippets deleted
"""
return await self.delete_many(where={"guild_id": guild_id})
upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Snippet
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 | 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. |