`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`	Find the first matching record using SQLModel select().
`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 SQLModel '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})

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

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

Python

"""
return await self.find_many(where={})

`_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 SQLModel 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 SQLModel (e.g., create, delete).

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

Constructs keyword arguments for SQLModel 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 SQLModel update operations.

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

Constructs keyword arguments for SQLModel 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 SQLModel upsert operations.

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

Find the first matching record using SQLModel select().

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

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) -> ModelType | None async ¶

find_unique(where: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType | 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[ModelType] async ¶

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

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

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

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

upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType 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` ¶

`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) -> ModelType | None` `async` ¶

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

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

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

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

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

`upsert(where: dict[str, Any], create: dict[str, Any], update: dict[str, Any], include: dict[str, bool] | None = None) -> ModelType` `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` ¶