`tux.database.controllers.guild` ¶

Classes:

Name	Description
`GuildController`	Controller for managing guild records.

Classes¶

`GuildController()` ¶

Bases: BaseController[Guild]

Controller for managing guild records.

This controller provides methods for managing guild records in the database. It inherits common CRUD operations from BaseController.

Initialize the GuildController with the guild table.

Methods:

Name	Description
`get_guild_by_id`	Get a guild by its ID.
`get_or_create_guild`	Get an existing guild or create it if it doesn't exist.
`insert_guild_by_id`	Insert a new guild.
`delete_guild_by_id`	Delete a guild by its ID.
`get_all_guilds`	Get all guilds.
`find_one`	Finds the first record matching specified criteria.
`find_unique`	Finds a single record by a unique constraint (e.g., ID).
`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/guild.py

Python

def __init__(self):
    """Initialize the GuildController with the guild table."""
    super().__init__("guild")
    # Type hint for better IDE support
    self.table: Any = self.table

Functions¶

`get_guild_by_id(guild_id: int) -> Guild | None` `async` ¶

Get a guild by its ID.

Parameters:

Name	Type	Description	Default
`guild_id`	`int`	The ID of the guild to get	required

Returns:

Type	Description
`Guild \| None`	The guild if found, None otherwise

Source code in tux/database/controllers/guild.py

Python

async def get_guild_by_id(self, guild_id: int) -> Guild | None:
    """Get a guild by its ID.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get

    Returns
    -------
    Guild | None
        The guild if found, None otherwise
    """
    return await self.find_one(where={"guild_id": guild_id})

`get_or_create_guild(guild_id: int) -> Guild` `async` ¶

Get an existing guild or create it if it doesn't exist.

Parameters:

Name	Type	Description	Default
`guild_id`	`int`	The ID of the guild to get or create	required

Returns:

Type	Description
`Guild`	The existing or newly created guild

Source code in tux/database/controllers/guild.py

Python

async def get_or_create_guild(self, guild_id: int) -> Guild:
    """Get an existing guild or create it if it doesn't exist.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to get or create

    Returns
    -------
    Guild
        The existing or newly created guild
    """
    return await self.table.upsert(
        where={"guild_id": guild_id},
        data={
            "create": {"guild_id": guild_id},
            "update": {},
        },
    )

`insert_guild_by_id(guild_id: int) -> Guild` `async` ¶

Insert a new guild.

Parameters:

Name	Type	Description	Default
`guild_id`	`int`	The ID of the guild to insert	required

Returns:

Type	Description
`Guild`	The created guild

Source code in tux/database/controllers/guild.py

Python

async def insert_guild_by_id(self, guild_id: int) -> Guild:
    """Insert a new guild.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to insert

    Returns
    -------
    Guild
        The created guild
    """
    return await self.create(data={"guild_id": guild_id})

`_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/guild.py

Python

async def delete_guild_by_id(self, guild_id: int) -> None:
    """Delete a guild by its ID.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete
    """
    await self.delete(where={"guild_id": guild_id})

async def get_all_guilds(self) -> list[Guild]:
    """Get all guilds.

    Returns
    -------
    list[Guild]
        List of all guilds
    """
    return await self.find_many(where={})

`delete_guild_by_id(guild_id: int) -> None` `async` ¶

Delete a guild by its ID.

Parameters:

Name	Type	Description	Default
`guild_id`	`int`	The ID of the guild to delete	required

Source code in tux/database/controllers/guild.py

Python

async def delete_guild_by_id(self, guild_id: int) -> None:
    """Delete a guild by its ID.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to delete
    """
    await self.delete(where={"guild_id": guild_id})

`get_all_guilds() -> list[Guild]` `async` ¶

Get all guilds.

Returns:

Type	Description
`list[Guild]`	List of all guilds

Source code in tux/database/controllers/guild.py

Python

async def get_all_guilds(self) -> list[Guild]:
    """Get all guilds.

    Returns
    -------
    list[Guild]
        List of all guilds
    """
    return await self.find_many(where={})

`_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.

`_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.

`_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).

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma create operations.

`_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.

`_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

Constructs keyword arguments for Prisma delete operations.

`_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.

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Guild | 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.

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | 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.

`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[Guild]` `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.

`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) -> Guild` `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) -> Guild | 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) -> Guild | 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) -> Guild` `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.

tux.database.controllers.guild ¶

Classes¶

GuildController() ¶

Functions¶

get_guild_by_id(guild_id: int) -> Guild | None async ¶

get_or_create_guild(guild_id: int) -> Guild async ¶

insert_guild_by_id(guild_id: int) -> Guild async ¶

_execute_query(operation: Callable[[], Any], error_msg: str) -> Any async ¶

delete_guild_by_id(guild_id: int) -> None async ¶

get_all_guilds() -> list[Guild] async ¶

_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None ¶

_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] ¶

_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

_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] ¶

_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any] ¶

find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Guild | None async ¶

find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None async ¶

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[Guild] async ¶

count(where: dict[str, Any]) -> int async ¶

create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Guild async ¶

update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None async ¶

delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None async ¶

upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Guild async ¶

update_many(where: dict[str, Any], data: dict[str, Any]) -> int async ¶

delete_many(where: dict[str, Any]) -> int async ¶

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

connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any] staticmethod ¶

safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod ¶

`tux.database.controllers.guild` ¶

`GuildController()` ¶

`get_guild_by_id(guild_id: int) -> Guild | None` `async` ¶

`get_or_create_guild(guild_id: int) -> Guild` `async` ¶

`insert_guild_by_id(guild_id: int) -> Guild` `async` ¶

`_execute_query(operation: Callable[[], Any], error_msg: str) -> Any` `async` ¶

`delete_guild_by_id(guild_id: int) -> None` `async` ¶

`get_all_guilds() -> list[Guild]` `async` ¶

`_add_include_arg_if_present(args: dict[str, Any], include: dict[str, bool] | None) -> None` ¶

`_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]` ¶

`_build_simple_args(key_name: str, key_value: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`_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]` ¶

`_build_upsert_args(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]` ¶

`find_one(where: dict[str, Any], include: dict[str, bool] | None = None, order: dict[str, str] | None = None) -> Guild | None` `async` ¶

`find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None` `async` ¶

`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[Guild]` `async` ¶

`count(where: dict[str, Any]) -> int` `async` ¶

`create(data: dict[str, Any], include: dict[str, bool] | None = None) -> Guild` `async` ¶

`update(where: dict[str, Any], data: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None` `async` ¶

`delete(where: dict[str, Any], include: dict[str, bool] | None = None) -> Guild | None` `async` ¶

`upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> Guild` `async` ¶

`update_many(where: dict[str, Any], data: dict[str, Any]) -> int` `async` ¶

`delete_many(where: dict[str, Any]) -> int` `async` ¶

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

`connect_or_create_relation(id_field: str, model_id: Any, create_data: dict[str, Any] | None = None) -> dict[str, Any]` `staticmethod` ¶

`safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any` `staticmethod` ¶