💽 Database usage

The bot uses SQLAlchemy ORM in asynchronous mode to work with the database. Documentation.

WARNING

Before you can use the database, you must add the use_db or require_db permission to the permissions list in your module's config.yaml file!

Declaring Models

Define your database models using SQLAlchemy's Declarative Mapping (e.g., inheriting from DeclarativeBase). It's common practice to place these in a separate file, like db.py.

# db.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer

class Base(DeclarativeBase):
    pass

class UserProfile(Base):
    __tablename__ = "user_profiles"

    user_id: Mapped[int] = mapped_column(primary_key=True)
    display_name: Mapped[str] = mapped_column(String(100))
    points: Mapped[int] = mapped_column(Integer, default=0)

# You can define other tables/models here too

Providing Metadata

Your main module class must expose the SQLAlchemy MetaData object associated with your models. This is done by overriding the db_meta property. The framework uses this metadata to automatically create the tables when the module loads (if they don't exist).

# main.py (inside your BaseModule class)
from .db_models import Base
from sqlalchemy import MetaData

# ... other imports and class definition ...

    @property
    def db_meta(self) -> MetaData:
        """Provides SQLAlchemy MetaData for table creation."""
        return Base.metadata

Database Initialization and Access

If database access is permitted (use_db or require_db in config) and the bot's database is enabled, the ModuleLoader will:

1. Create a Database object (base.db.Database) specifically for your module. By default, this might be a separate SQLite file (e.g., ./ModuleName/module_db.sqlite) or a schema/database in a shared RDBMS, depending on the bot's config.db_url.
2. Assign this Database object to the self.db attribute of your module instance.
3. Use self.db_meta to run metadata.create_all(bind=engine) asynchronously.
4. Call the async def on_db_ready(self) method in your module after the tables are ensured to exist.

The self.db object contains:

• engine: An AsyncEngine instance connected to your module's specific database/schema.
• session_maker: An async_sessionmaker bound to the engine, configured with expire_on_commit=False. Use this to create asynchronous sessions.

Performing Database Operations

Use the async_sessionmaker (self.db.session_maker) to create sessions within an async with block for database operations.

# main.py (inside your BaseModule class)
from .db_models import UserProfile
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update

# ... other methods ...

    async def get_user_points(self, user_id: int) -> int:
        """Fetches user points from the database."""
        async with self.db.session_maker() as session: # Create a session
            session: AsyncSession # Type hint for clarity
            # Use scalar_one_or_none for fetching a single optional record's column
            points = await session.scalar(
                select(UserProfile.points).where(UserProfile.user_id == user_id)
            )
            return points if points is not None else 0

    async def add_points(self, user_id: int, points_to_add: int):
        """Adds points to a user, creating profile if needed."""
        async with self.db.session_maker() as session:
            # Try to update existing user
            stmt = (
                update(UserProfile)
                .where(UserProfile.user_id == user_id)
                .values(points=UserProfile.points + points_to_add)
            )
            result = await session.execute(stmt)

            # If no rows were updated, user doesn't exist yet
            if result.rowcount == 0:
                # Create a new profile (assuming display_name can be fetched elsewhere)
                # In a real scenario, you might fetch the name or handle this differently
                display_name = f"User_{user_id}" # Placeholder
                new_profile = UserProfile(
                    user_id=user_id,
                    display_name=display_name,
                    points=points_to_add
                )
                session.add(new_profile)

            # Commit changes for this session
            await session.commit()

    async def on_db_ready(self):
        """Called after DB is set up. Can be used for initial data loading."""
        self.logger.info("Database is ready! Performing initial checks...")
        # Example: Log the number of existing user profiles
        async with self.db.session_maker() as session:
            count = await session.scalar(select(func.count(UserProfile.user_id)))
            self.logger.info(f"Found {count} user profiles in the database.")

Database Migrations

For managing changes to your database schema across module updates, place migration scripts in a db_migrations/ directory within your module. See base.db_migration.DBMigration and the ModuleManager.update_from_git method for how these are potentially applied during updates based on version comparison.