Skip to content

tux.database.controllers.guild_config

Classes:

Name Description
GuildConfigController

Controller for managing GuildConfig records.

Classes

GuildConfigController()

Bases: BaseController[GuildConfig]

Controller for managing GuildConfig records.

Initializes the BaseController for a specific table.

Parameters:

Name Type Description Default
table type[ModelType]

The SQLModel model class representing the table.

 required
session AsyncSession

An optional SQLAlchemy async session to use for database operations.

 None

Methods:

Name Description
get_field

Fetch a single scalar field from GuildConfig.
get_perm_level_role

Get the role id for a specific permission level.
get_perm_level_roles

Get the role ids for all permission levels from lower_bound up to 7.
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_config.py
Python
def __init__(self) -> None:
    super().__init__(GuildConfig)

Functions

get_field(guild_id: int, field: str) -> Any async

Fetch a single scalar field from GuildConfig. Returns None if the record or field does not exist.

Source code in tux/database/controllers/guild_config.py
Python
async def get_field(self, guild_id: int, field: str) -> Any:
    """
    Fetch a single scalar field from GuildConfig.
    Returns None if the record or field does not exist.
    """
    cfg = await self.find_unique(where={"guild_id": guild_id})
    if cfg is None:
        logger.warning(f"No GuildConfig found for guild_id={guild_id}")
        return None
    value = self.safe_get_attr(cfg, field, None)
    logger.debug(f"GuildConfig[{guild_id}].{field} = {value!r}")
    return value
get_perm_level_role(guild_id: int, level: int) -> int | None async

Get the role id for a specific permission level.

Source code in tux/database/controllers/guild_config.py
Python
async def get_perm_level_role(self, guild_id: int, level: int) -> int | None:
    """
    Get the role id for a specific permission level.
    """
    field = f"perm_level_{level}_role_id"
    try:
        return await self.get_field(guild_id, field)
    except Exception as e:
        logger.error(f"Error getting perm level role {level} for guild {guild_id}: {e}")
        return None
get_perm_level_roles(guild_id: int, lower_bound: int) -> list[int] | None async

Get the role ids for all permission levels from lower_bound up to 7.

Source code in tux/database/controllers/guild_config.py
Python
async def get_perm_level_roles(self, guild_id: int, lower_bound: int) -> list[int] | None:
    """
    Get the role ids for all permission levels from lower_bound up to 7.
    """
    try:
        role_ids: list[int] = []
        for lvl in range(lower_bound, 8):
            field = f"perm_level_{lvl}_role_id"
            rid = await self.get_field(guild_id, field)
            if rid is not None:
                role_ids.append(rid)
        logger.debug(f"Retrieved perm roles >={lower_bound} for guild {guild_id}: {role_ids}")
    except Exception as e:
        logger.error(f"Error getting perm level roles for guild {guild_id}: {e}")
        return None
    else:
        return role_ids
_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_config.py
Python
async def get_join_log_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "join_log_id")

async def get_private_log_id(self, guild_id: int) -> int | 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 SQLModel find operations.

Source code in tux/database/controllers/guild_config.py
Python
async def get_report_log_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "report_log_id")

async def get_dev_log_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "dev_log_id")

async def get_jail_channel_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "jail_channel_id")

async def get_general_channel_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "general_channel_id")

async def get_starboard_channel_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "starboard_channel_id")

async def get_base_staff_role_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "base_staff_role_id")

async def get_base_member_role_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "base_member_role_id")
_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).

Source code in tux/database/controllers/guild_config.py
Python
async def get_jail_role_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "jail_role_id")

async def get_quarantine_role_id(self, guild_id: int) -> int | None:
    return await self.get_field(guild_id, "quarantine_role_id")

# --- Generic Field Upserter ---

async def _upsert_field(self, guild_id: int, field: str, value: Any) -> GuildConfig:
    """
_upsert_field(guild_id: int, field: str, value: Any) -> GuildConfig async

Upsert a single scalar field on GuildConfig, ensuring the GuildConfig row (and its guild relation) exists.

Source code in tux/database/controllers/guild_config.py
Python
async def _upsert_field(self, guild_id: int, field: str, value: Any) -> GuildConfig:
    """
    Upsert a single scalar field on GuildConfig, ensuring the
    GuildConfig row (and its guild relation) exists.
    """
    return await self.upsert(
        where={"guild_id": guild_id},
        create={
            field: value,
            "guild": self.connect_or_create_relation("guild_id", guild_id),
        },
        update={field: value},
    )
_build_create_args(data: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel create operations.

Source code in tux/database/controllers/guild_config.py
Python
GuildConfig row (and its guild relation) exists.
"""
return await self.upsert(
    where={"guild_id": guild_id},
    create={
        field: value,
        "guild": self.connect_or_create_relation("guild_id", guild_id),
_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.

Source code in tux/database/controllers/guild_config.py
Python
        update={field: value},
    )

# --- Write Methods ---

async def update_guild_prefix(self, guild_id: int, prefix: str) -> GuildConfig:
    return await self._upsert_field(guild_id, "prefix", prefix)

async def update_mod_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "mod_log_id", channel_id)
_build_delete_args(where: dict[str, Any], include: dict[str, bool] | None = None) -> dict[str, Any]

Constructs keyword arguments for SQLModel delete operations.

Source code in tux/database/controllers/guild_config.py
Python
async def update_audit_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "audit_log_id", channel_id)

async def update_join_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "join_log_id", channel_id)

async def update_private_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
_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.

Source code in tux/database/controllers/guild_config.py
Python
async def update_report_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "report_log_id", channel_id)

async def update_dev_log_id(self, guild_id: int, channel_id: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "dev_log_id", channel_id)

async def update_jail_channel_id(self, guild_id: int, cid: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "jail_channel_id", cid)

async def update_general_channel_id(self, guild_id: int, cid: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "general_channel_id", cid)

async def update_starboard_channel_id(self, guild_id: int, cid: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "starboard_channel_id", cid)

async def update_base_staff_role_id(self, guild_id: int, rid: int) -> GuildConfig:
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().

Source code in tux/database/controllers/guild_config.py
Python
    return await self._upsert_field(guild_id, "base_member_role_id", rid)

async def update_jail_role_id(self, guild_id: int, rid: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "jail_role_id", rid)

async def update_quarantine_role_id(self, guild_id: int, rid: int) -> GuildConfig:
    return await self._upsert_field(guild_id, "quarantine_role_id", rid)

async def update_perm_level_role(self, guild_id: int, level: int, role_id: int) -> GuildConfig:
    field = f"perm_level_{level}_role_id"
    return await self._upsert_field(guild_id, field, role_id)

async def update_guild_config(self, guild_id: int, data: dict[str, Any]) -> GuildConfig | None:
    return await self.update(where={"guild_id": guild_id}, data=data)
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.
Source code in tux/database/controllers/guild_config.py
Python
async def delete_guild_config(self, guild_id: int) -> GuildConfig | None:
    return await self.delete(where={"guild_id": guild_id})

async def delete_guild_prefix(self, guild_id: int) -> GuildConfig | None:
    return await self.update(where={"guild_id": guild_id}, data={"prefix": None})
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.