Skip to content

tux.database.controllers.guild_config

Classes:

Name Description
GuildConfigController

Controller that manages per-guild configuration settings.

Classes

GuildConfigController()

Bases: BaseController[GuildConfig]

Controller that manages per-guild configuration settings.

Methods:

Name Description
ensure_guild_exists

Ensure the guild exists in the database.

insert_guild_config

Insert a new guild config into the database.

get_guild_config

Get a guild config from the database.

get_guild_prefix

Get a guild prefix from the database.

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 the lower_bound up to but not including 8.

find_one

Return the first row that matches where or None.

find_many

Return a list of rows matching where (or all rows).

execute_transaction

Execute callback inside a database session / transaction block.

safe_get_attr

Return getattr(obj, attr, default) - keeps old helper available.

connect_or_create_relation

Return a dict with a single key that can be merged into data dicts.

Source code in tux/database/controllers/guild_config.py
Python
def __init__(self):
    super().__init__(GuildConfig)
    self._guild_repo: BaseController[Guild] = BaseController(Guild)

Functions

ensure_guild_exists(guild_id: int) -> Any async

Ensure the guild exists in the database.

Source code in tux/database/controllers/guild_config.py
Python
async def ensure_guild_exists(self, guild_id: int) -> Any:
    """Ensure the guild exists in the database."""
    guild: Any = await self._guild_repo.find_one(where={"guild_id": guild_id})
    if guild is None:
        return await self._guild_repo.create(data={"guild_id": guild_id})
    return guild
insert_guild_config(guild_id: int) -> Any async

Insert a new guild config into the database.

Source code in tux/database/controllers/guild_config.py
Python
async def insert_guild_config(self, guild_id: int) -> Any:
    """Insert a new guild config into the database."""
    await self.ensure_guild_exists(guild_id)
    return await self.create(data={"guild_id": guild_id})
get_guild_config(guild_id: int) -> Any async

Get a guild config from the database.

Source code in tux/database/controllers/guild_config.py
Python
async def get_guild_config(self, guild_id: int) -> Any:
    """Get a guild config from the database."""
    return await self.find_one(where={"guild_id": guild_id})
get_guild_prefix(guild_id: int) -> str | None async

Get a guild prefix from the database.

Source code in tux/database/controllers/guild_config.py
Python
async def get_guild_prefix(self, guild_id: int) -> str | None:
    """Get a guild prefix from the database."""
    config: Any = await self.find_one(where={"guild_id": guild_id})
    return None if config is None else config.prefix
_execute_query(op: Callable[[AsyncSession], Any], span_desc: str) -> Any async

Run op inside a managed session & sentry span (if enabled).

Source code in tux/database/controllers/guild_config.py
Python
    log_channel_ids: dict[str, GuildConfigScalarFieldKeys] = {
        "mod": "mod_log_id",
        "audit": "audit_log_id",
        "join": "join_log_id",
        "private": "private_log_id",
        "report": "report_log_id",
        "dev": "dev_log_id",
    }
    return await self.get_guild_config_field_value(guild_id, log_channel_ids[log_type])

async def get_perm_level_role(self, guild_id: int, level: str) -> int | None:
    """
    Get the role id for a specific permission level.
    """
    try:
        role_id = await self.get_guild_config_field_value(guild_id, level)  # type: ignore
        logger.debug(f"Retrieved role_id {role_id} for guild {guild_id} and level {level}")
    except Exception as e:
        logger.error(f"Error getting perm level role: {e}")
get_perm_level_role(guild_id: int, level: str) -> 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: str) -> int | None:
    """
    Get the role id for a specific permission level.
    """
    try:
        role_id = await self.get_guild_config_field_value(guild_id, level)  # type: ignore
        logger.debug(f"Retrieved role_id {role_id} for guild {guild_id} and level {level}")
    except Exception as e:
        logger.error(f"Error getting perm level role: {e}")
        return None
    return role_id
get_perm_level_roles(guild_id: int, lower_bound: int) -> list[int] | None async

Get the role ids for all permission levels from the lower_bound up to but not including 8.

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 the lower_bound up to but not including 8.
    """
    perm_level_roles: dict[int, str] = {
        0: "perm_level_0_role_id",
        1: "perm_level_1_role_id",
        2: "perm_level_2_role_id",
        3: "perm_level_3_role_id",
        4: "perm_level_4_role_id",
        5: "perm_level_5_role_id",
        6: "perm_level_6_role_id",
        7: "perm_level_7_role_id",
    }

    try:
        role_ids: list[int] = []

        for level in range(lower_bound, 8):
            if role_field := perm_level_roles.get(level):
                role_id = await self.get_guild_config_field_value(guild_id, role_field)  # type: ignore

                if role_id:
                    role_ids.append(role_id)

        logger.debug(f"Retrieved role_ids {role_ids} for guild {guild_id} with lower bound {lower_bound}")

    except Exception as e:
        logger.error(f"Error getting perm level roles: {e}")
        return None

    return role_ids
find_one(*, where: dict[str, Any], include: dict[str, bool] | None = None, **__: Any) -> ModelT | None async

Return the first row that matches where or None.

Source code in tux/database/controllers/guild_config.py
Python
Get the role ids for all permission levels from the lower_bound up to but not including 8.
"""
perm_level_roles: dict[int, str] = {
    0: "perm_level_0_role_id",
    1: "perm_level_1_role_id",
    2: "perm_level_2_role_id",
    3: "perm_level_3_role_id",
    4: "perm_level_4_role_id",
    5: "perm_level_5_role_id",
    6: "perm_level_6_role_id",
    7: "perm_level_7_role_id",
}

try:
    role_ids: list[int] = []
find_many(*, where: dict[str, Any] | None = None, include: dict[str, bool] | None = None, order: dict[str, str] | None = None, take: int | None = None, skip: int | None = None) -> list[ModelT] async

Return a list of rows matching where (or all rows).

Source code in tux/database/controllers/guild_config.py
Python
    return role_ids

async def get_guild_config_field_value(
    self,
    guild_id: int,
    field: GuildConfigScalarFieldKeys,
) -> Any:
    config: Any = await self.find_one(where={"guild_id": guild_id})

    if config is None:
        logger.warning(f"No guild config found for guild_id: {guild_id}")
        return None

    value = getattr(config, field, None)

    logger.debug(f"Retrieved field value for {field}: {value}")

    return value

async def get_mod_log_id(self, guild_id: int) -> int | None:
    return await self.get_guild_config_field_value(guild_id, "mod_log_id")

async def get_audit_log_id(self, guild_id: int) -> int | None:
    return await self.get_guild_config_field_value(guild_id, "audit_log_id")

async def get_join_log_id(self, guild_id: int) -> int | None:
    return await self.get_guild_config_field_value(guild_id, "join_log_id")
execute_transaction(callback: Callable[[], Any]) -> Any async

Execute callback inside a database session / transaction block.

Source code in tux/database/controllers/guild_config.py
Python
    audit_log_id: int,
) -> Any:
    await self.ensure_guild_exists(guild_id)

    return await self.upsert(
        where={"guild_id": guild_id},
        data={
            "create": {
                "guild_id": guild_id,
                "audit_log_id": audit_log_id,
            },
            "update": {"audit_log_id": audit_log_id},
        },
safe_get_attr(obj: Any, attr: str, default: Any = None) -> Any staticmethod

Return getattr(obj, attr, default) - keeps old helper available.

Source code in tux/database/controllers/guild_config.py
Python
    join_log_id: int,
) -> Any:
    await self.ensure_guild_exists(guild_id)
connect_or_create_relation(id_field: str, model_id: Any, *_: Any, **__: Any) -> dict[str, Any] staticmethod

Return a dict with a single key that can be merged into data dicts.

The calling code does something like::

Text Only
data = {"guild": connect_or_create_relation("guild_id", guild_id)}

We map that pattern to a very small helper that collapses to {"guild_id": guild_id}.

Source code in tux/database/controllers/guild_config.py
Python
                "guild_id": guild_id,
                "join_log_id": join_log_id,
            },
            "update": {"join_log_id": join_log_id},
        },
    )

async def update_private_log_id(
    self,
    guild_id: int,
    private_log_id: int,