core ¶

Core module for Tux bot.

This module provides the core infrastructure including: - Base cog class for extensions - Database service for data persistence

Modules:

app –

Tux application entrypoint and lifecycle management.
base_cog –

Enhanced base cog with database access and automatic usage generation.
bot –

Tux Discord bot core implementation.
checks –

Dynamic Permission System - Fully Database-Driven.
cog_loader –

Dynamic cog loading system with priority-based ordering and telemetry.
context –

Command and Interaction Context Utilities.
converters –

Discord.py command converters for Tux bot.
decorators –

Dynamic Permission Decorators.
flags –

Flag converters for Discord bot commands.
logging –

Centralized Loguru Configuration for Tux Discord Bot.
permission_system –

Dynamic permission system for guild-specific permission hierarchies.
prefix_manager –

Prefix management with in-memory caching for optimal performance.
setup –

Setup services for bot initialization.
task_monitor –

Task monitoring and cleanup utilities for the Tux bot.
types –

Type definitions for Tux core components.

Classes:

BaseCog –

Enhanced base cog class providing database access and automatic usage generation.
DatabaseService –

Async database service for PostgreSQL.

Classes¶

BaseCog ¶

Python

BaseCog(bot: Tux)

Bases: Cog

Enhanced base cog class providing database access and automatic usage generation.

This class serves as the foundation for all bot cogs, offering convenient access to database controllers, configuration values, and automatic command usage string generation based on function signatures.

Attributes:

bot (Tux) –

The bot instance this cog is attached to.
_unload_task (Task[None] | None) –

Background task for graceful cog unloading when config is missing.

Notes

All cogs should inherit from this class to gain access to: - Database operations via self.db - Configuration access via self.get_config() - Automatic command usage generation - Graceful unloading on missing configuration

Initialize the base cog with bot instance and command usage setup.

Parameters:

bot (Tux) –

The bot instance this cog will be attached to.

Notes

Automatically generates usage strings for all commands in this cog that don't have explicit usage strings defined.

Subclassed by:

Methods:

get_config –

Get a configuration value from CONFIG with support for nested keys.
__repr__ –

Return a string representation of the cog instance.
unload_if_missing_config –

Check if required configuration is missing and log warning.

Attributes¶

db `property` ¶

Python

db: DatabaseCoordinator

Get the database coordinator for accessing database controllers.

Returns:

DatabaseCoordinator –

Coordinator providing access to all database controllers.

Examples:

Python Console Session

>>> await self.db.guild_config.get_guild_config(guild_id)
>>> await self.db.cases.create_case(...)

Notes

This property provides convenient access to database operations without needing to access self.bot.db directly.

Functions¶

_setup_command_usage ¶

Python

_setup_command_usage() -> None

Generate usage strings for all commands in this cog that lack explicit usage.

The generated usage follows the pattern: <qualified_name> <param tokens>

Where: - Required parameters are denoted as <name: Type> - Optional parameters are denoted as [name: Type] - The prefix is intentionally omitted (provided by ctx.prefix)

Examples:

ban <member: Member> [reason: str] config set <key: str> <value: str>

Notes

Respects explicit usage strings if already set on a command. Errors during generation are logged but don't prevent cog loading.

_generate_usage ¶

Python

_generate_usage(command: Command[Any, ..., Any]) -> str

Generate a usage string with support for flags and positional parameters.

This method inspects the command's callback signature to detect: - FlagConverter parameters (e.g., --flag value) - Positional parameters (e.g., <required> or [optional])

Parameters:

command (Command) –

The command to generate usage for.

Returns:

str –

Generated usage string, or qualified command name as fallback.

Notes

Delegates to shared usage generator for consistency across all cogs. Falls back gracefully to command name if generation fails.

get_config ¶

Python

get_config(key: str, default: Any = None) -> Any

Get a configuration value from CONFIG with support for nested keys.

Parameters:

key (str) –

The configuration key to retrieve. Supports dot notation for nested values (e.g., "BOT_INFO.BOT_NAME").
default (Any, default: None ) –

Default value to return if key is not found, by default None.

Returns:

Any –

The configuration value or default if not found.

Examples:

Python Console Session

>>> self.get_config("BOT_INFO.BOT_NAME")
'Tux'
>>> self.get_config("MISSING_KEY", "fallback")
'fallback'

Notes

Errors during retrieval are logged but don't raise exceptions. Returns the default value on any error.

repr ¶

Python

__repr__() -> str

Return a string representation of the cog instance.

Returns:

str –

String representation in format <CogName bot=BotUser>.

unload_if_missing_config ¶

Python

unload_if_missing_config(condition: bool, config_name: str) -> bool

Check if required configuration is missing and log warning.

This allows cogs to detect missing configuration at load time and return early from init to prevent partial initialization.

Parameters:

condition (bool) –

True if config is missing (should unload), False otherwise.
config_name (str) –

Name of the missing configuration for logging purposes.

Returns:

bool –

True if config is missing (caller should return early), False if config is present.

Examples:

Python Console Session

>>> def __init__(self, bot: Tux):
...     super().__init__(bot)
...     if self.unload_if_missing_config(not CONFIG.GITHUB_TOKEN, "GITHUB_TOKEN"):
...         return  # Exit early, cog will be partially loaded but won't register commands
...     self.github_client = GitHubClient()

Notes

When this returns True, the cog's init should return early to avoid initializing services that depend on the missing config. The cog will be loaded but commands won't be registered properly, preventing runtime errors.

For complete cog unloading, the bot owner should remove the cog from the modules directory or use the reload system to unload it programmatically.

_unload_self `async` ¶

Python

_unload_self(extension_name: str) -> None

Perform the actual cog unload operation.

Parameters:

extension_name (str) –

Full extension name to unload.

Notes

This is called as a background task by unload_if_missing_config(). Errors during unload are logged but don't raise exceptions.

DatabaseService ¶

Python

DatabaseService(echo: bool = False)

Async database service for PostgreSQL.

Provides connection management, session handling, query execution with retry logic, and health checks for the PostgreSQL database.

Attributes:

_engine (AsyncEngine | None) –

SQLAlchemy async engine for database connections.
_session_factory (async_sessionmaker[AsyncSession] | None) –

Factory for creating database sessions.
_echo (bool) –

Whether to log SQL queries (useful for debugging).

Initialize the database service.

Parameters:

echo (bool, default: False ) –

Whether to enable SQL query logging (default is False).

Returned by:

Used by:

Methods:

connect –

Connect to the PostgreSQL database.
disconnect –

Disconnect from the database and dispose of the connection pool.
is_connected –

Check if database is currently connected.
test_connection –

Test database connectivity with a simple query.
session –

Get a database session context manager.
execute_transaction –

Execute a callback inside a database transaction.
execute_query –

Execute database operation with automatic retry logic.
health_check –

Perform database health check.
validate_schema –

Validate that the database schema matches the current model definitions.

Attributes¶

engine `property` ¶

Python

engine: AsyncEngine | None

Get the database engine.

Returns:

AsyncEngine | None –

The SQLAlchemy async engine, or None if not connected.

Notes

Primarily used for testing and advanced operations.

Functions¶

connect `async` ¶

Python

connect(database_url: str, **kwargs: Any) -> None

Connect to the PostgreSQL database.

Parameters:

database_url (str) –

PostgreSQL connection URL in format: postgresql+psycopg://user:password@host:port/database
**kwargs (Any, default: {} ) –

Additional arguments passed to create_async_engine.

disconnect `async` ¶

Python

disconnect() -> None

Disconnect from the database and dispose of the connection pool.

is_connected ¶

Python

is_connected() -> bool

Check if database is currently connected.

Returns:

bool –

True if connected, False otherwise.

test_connection `async` ¶

Python

test_connection() -> None

Test database connectivity with a simple query.

Raises:

Exception –

If the database connection fails or the test query fails.

session `async` ¶

Python

session() -> AsyncGenerator[AsyncSession]

Get a database session context manager.

Automatically handles connection, commit, and rollback.

Yields:

AsyncSession –

An active database session.

Examples:

Python Console Session

>>> async with db.session() as session:
...     result = await session.execute(select(User))
...     users = result.scalars().all()

execute_transaction `async` ¶

Python

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

Execute a callback inside a database transaction.

Parameters:

callback (Callable[[], Any]) –

Async function to execute within the transaction.

Returns:

Any –

The return value of the callback function.

Notes

If the transaction fails, it will be rolled back automatically.

execute_query `async` ¶

Python

execute_query(operation: Callable[[AsyncSession], Awaitable[T]], span_desc: str) -> T

Execute database operation with automatic retry logic.

Parameters:

operation (Callable[[AsyncSession], Awaitable[T]]) –

Async function that performs database operations.
span_desc (str) –

Description for Sentry performance monitoring.

Returns:

T –

Result of the operation.

Notes

Retries the operation automatically on transient failures.

_execute_with_retry `async` ¶

Python

_execute_with_retry(
    operation: Callable[[AsyncSession], Awaitable[T]],
    span_desc: str,
    max_retries: int = 3,
    backoff_factor: float = 0.5,
) -> T

Execute database operation with exponential backoff retry logic.

Parameters:

operation (Callable[[AsyncSession], Awaitable[T]]) –

Database operation to execute.
span_desc (str) –

Description for monitoring/logging.
max_retries (int, default: 3 ) –

Maximum number of retry attempts (default is 3).
backoff_factor (float, default: 0.5 ) –

Multiplier for exponential backoff (default is 0.5).

Returns:

T –

Result of the operation.

Raises:

TimeoutError –

If the operation times out after all retries.
DisconnectionError –

If database disconnection occurs after all retries.
OperationalError –

If database operational error occurs after all retries.
RuntimeError –

If the retry loop completes unexpectedly without return or exception.

health_check `async` ¶

Python

health_check() -> dict[str, Any]

Perform database health check.

Returns:

dict[str, Any] –

Health check result with status and optional error message. Status can be: "healthy", "unhealthy", or "disconnected".

Examples:

Python Console Session

>>> result = await db.health_check()
>>> print(result)
{'status': 'healthy', 'mode': 'async'}

validate_schema `async` ¶

Python

validate_schema() -> dict[str, Any]

Validate that the database schema matches the current model definitions.

Uses SQLAlchemy's metadata reflection to compare the actual database schema with the defined model metadata. Much more efficient and accurate than manual queries.

Returns:

dict[str, Any] –

Schema validation result with status and optional error message. Status can be: "valid", "invalid", or "error".

Examples:

Python Console Session

>>> result = await db.validate_schema()
>>> print(result)
{'status': 'valid', 'mode': 'async'}

core ¶

Classes¶

BaseCog ¶

Attributes¶

db property ¶

Functions¶

_setup_command_usage ¶

_generate_usage ¶

get_config ¶

__repr__ ¶

unload_if_missing_config ¶

_unload_self async ¶

DatabaseService ¶

Attributes¶

engine property ¶

Functions¶

connect async ¶

disconnect async ¶

is_connected ¶

test_connection async ¶

session async ¶

execute_transaction async ¶

execute_query async ¶

_execute_with_retry async ¶

health_check async ¶

validate_schema async ¶

db `property` ¶

repr ¶

_unload_self `async` ¶

engine `property` ¶

connect `async` ¶

disconnect `async` ¶

test_connection `async` ¶

session `async` ¶

execute_transaction `async` ¶

execute_query `async` ¶

_execute_with_retry `async` ¶

health_check `async` ¶

validate_schema `async` ¶