API Reference¶

Complete API documentation for all chapkit modules, classes, and functions.

Core Layer¶

Framework-agnostic infrastructure components.

Database¶

`database` ¶

Async SQLAlchemy database connection manager.

`Database` ¶

Generic async SQLAlchemy database connection manager.

Source code in src/chapkit/core/database.py

class Database:
    """Generic async SQLAlchemy database connection manager."""

    def __init__(
        self,
        url: str,
        *,
        echo: bool = False,
        alembic_dir: Path | None = None,
        auto_migrate: bool = True,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_recycle: int = 3600,
        pool_pre_ping: bool = True,
    ) -> None:
        """Initialize database with connection URL and pool configuration."""
        self.url = url
        self.alembic_dir = alembic_dir
        self.auto_migrate = auto_migrate

        # Build engine kwargs - skip pool params for in-memory SQLite databases
        engine_kwargs: dict = {"echo": echo, "future": True}
        if ":memory:" not in url:
            # Only add pool params for non-in-memory databases
            engine_kwargs.update(
                {
                    "pool_size": pool_size,
                    "max_overflow": max_overflow,
                    "pool_recycle": pool_recycle,
                    "pool_pre_ping": pool_pre_ping,
                }
            )

        self.engine: AsyncEngine = create_async_engine(url, **engine_kwargs)
        self._session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
            bind=self.engine, class_=AsyncSession, expire_on_commit=False
        )

    async def init(self) -> None:
        """Initialize database tables using Alembic migrations or direct creation."""
        import asyncio

        # Import Base here to avoid circular import at module level
        from chapkit.core.models import Base

        # For databases without migrations, use direct table creation
        if not self.auto_migrate:
            async with self.engine.begin() as conn:
                await conn.run_sync(Base.metadata.create_all)
        else:
            # Use Alembic migrations
            alembic_cfg = Config()

            # Use custom alembic directory if provided, otherwise use bundled migrations
            if self.alembic_dir is not None:
                alembic_cfg.set_main_option("script_location", str(self.alembic_dir))
            else:
                alembic_cfg.set_main_option(
                    "script_location", str(Path(__file__).parent.parent.parent.parent / "alembic")
                )

            alembic_cfg.set_main_option("sqlalchemy.url", self.url)

            # Run upgrade in executor to avoid event loop conflicts
            loop = asyncio.get_running_loop()
            await loop.run_in_executor(None, command.upgrade, alembic_cfg, "head")

    @asynccontextmanager
    async def session(self) -> AsyncIterator[AsyncSession]:
        """Create a database session context manager."""
        async with self._session_factory() as s:
            yield s

    async def dispose(self) -> None:
        """Dispose of database engine and connection pool."""
        await self.engine.dispose()

`init(url, *, echo=False, alembic_dir=None, auto_migrate=True, pool_size=5, max_overflow=10, pool_recycle=3600, pool_pre_ping=True)` ¶

Initialize database with connection URL and pool configuration.

Source code in src/chapkit/core/database.py

def __init__(
    self,
    url: str,
    *,
    echo: bool = False,
    alembic_dir: Path | None = None,
    auto_migrate: bool = True,
    pool_size: int = 5,
    max_overflow: int = 10,
    pool_recycle: int = 3600,
    pool_pre_ping: bool = True,
) -> None:
    """Initialize database with connection URL and pool configuration."""
    self.url = url
    self.alembic_dir = alembic_dir
    self.auto_migrate = auto_migrate

    # Build engine kwargs - skip pool params for in-memory SQLite databases
    engine_kwargs: dict = {"echo": echo, "future": True}
    if ":memory:" not in url:
        # Only add pool params for non-in-memory databases
        engine_kwargs.update(
            {
                "pool_size": pool_size,
                "max_overflow": max_overflow,
                "pool_recycle": pool_recycle,
                "pool_pre_ping": pool_pre_ping,
            }
        )

    self.engine: AsyncEngine = create_async_engine(url, **engine_kwargs)
    self._session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
        bind=self.engine, class_=AsyncSession, expire_on_commit=False
    )

`init()` `async` ¶

Initialize database tables using Alembic migrations or direct creation.

Source code in src/chapkit/core/database.py

async def init(self) -> None:
    """Initialize database tables using Alembic migrations or direct creation."""
    import asyncio

    # Import Base here to avoid circular import at module level
    from chapkit.core.models import Base

    # For databases without migrations, use direct table creation
    if not self.auto_migrate:
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
    else:
        # Use Alembic migrations
        alembic_cfg = Config()

        # Use custom alembic directory if provided, otherwise use bundled migrations
        if self.alembic_dir is not None:
            alembic_cfg.set_main_option("script_location", str(self.alembic_dir))
        else:
            alembic_cfg.set_main_option(
                "script_location", str(Path(__file__).parent.parent.parent.parent / "alembic")
            )

        alembic_cfg.set_main_option("sqlalchemy.url", self.url)

        # Run upgrade in executor to avoid event loop conflicts
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, command.upgrade, alembic_cfg, "head")

`session()` `async` ¶

Create a database session context manager.

Source code in src/chapkit/core/database.py

@asynccontextmanager
async def session(self) -> AsyncIterator[AsyncSession]:
    """Create a database session context manager."""
    async with self._session_factory() as s:
        yield s

`dispose()` `async` ¶

Dispose of database engine and connection pool.

Source code in src/chapkit/core/database.py

async def dispose(self) -> None:
    """Dispose of database engine and connection pool."""
    await self.engine.dispose()

`SqliteDatabase` ¶

Bases: Database

SQLite-specific database implementation with optimizations.

Source code in src/chapkit/core/database.py

class SqliteDatabase(Database):
    """SQLite-specific database implementation with optimizations."""

    def __init__(
        self,
        url: str,
        *,
        echo: bool = False,
        alembic_dir: Path | None = None,
        auto_migrate: bool = True,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_recycle: int = 3600,
        pool_pre_ping: bool = True,
    ) -> None:
        """Initialize SQLite database with connection URL and pool configuration."""
        self.url = url
        self.alembic_dir = alembic_dir
        self.auto_migrate = auto_migrate

        # Build engine kwargs - pool params only for non-in-memory databases
        engine_kwargs: dict = {"echo": echo, "future": True}
        if not self._is_in_memory_url(url):
            # File-based databases can use pool configuration
            engine_kwargs.update(
                {
                    "pool_size": pool_size,
                    "max_overflow": max_overflow,
                    "pool_recycle": pool_recycle,
                    "pool_pre_ping": pool_pre_ping,
                }
            )

        self.engine: AsyncEngine = create_async_engine(url, **engine_kwargs)
        _install_sqlite_connect_pragmas(self.engine)
        self._session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
            bind=self.engine, class_=AsyncSession, expire_on_commit=False
        )

    @staticmethod
    def _is_in_memory_url(url: str) -> bool:
        """Check if URL represents an in-memory database."""
        return ":memory:" in url

    def is_in_memory(self) -> bool:
        """Check if this is an in-memory database."""
        return self._is_in_memory_url(self.url)

    async def init(self) -> None:
        """Initialize database tables and configure SQLite using Alembic migrations."""
        # Import Base here to avoid circular import at module level
        from chapkit.core.models import Base

        # Set WAL mode first (if not in-memory)
        if not self.is_in_memory():
            async with self.engine.begin() as conn:
                await conn.exec_driver_sql("PRAGMA journal_mode=WAL;")

        # For in-memory databases or when migrations are disabled, use direct table creation
        if self.is_in_memory() or not self.auto_migrate:
            async with self.engine.begin() as conn:
                await conn.run_sync(Base.metadata.create_all)
        else:
            # For file-based databases, use Alembic migrations
            await super().init()

`init(url, *, echo=False, alembic_dir=None, auto_migrate=True, pool_size=5, max_overflow=10, pool_recycle=3600, pool_pre_ping=True)` ¶

Initialize SQLite database with connection URL and pool configuration.

Source code in src/chapkit/core/database.py

def __init__(
    self,
    url: str,
    *,
    echo: bool = False,
    alembic_dir: Path | None = None,
    auto_migrate: bool = True,
    pool_size: int = 5,
    max_overflow: int = 10,
    pool_recycle: int = 3600,
    pool_pre_ping: bool = True,
) -> None:
    """Initialize SQLite database with connection URL and pool configuration."""
    self.url = url
    self.alembic_dir = alembic_dir
    self.auto_migrate = auto_migrate

    # Build engine kwargs - pool params only for non-in-memory databases
    engine_kwargs: dict = {"echo": echo, "future": True}
    if not self._is_in_memory_url(url):
        # File-based databases can use pool configuration
        engine_kwargs.update(
            {
                "pool_size": pool_size,
                "max_overflow": max_overflow,
                "pool_recycle": pool_recycle,
                "pool_pre_ping": pool_pre_ping,
            }
        )

    self.engine: AsyncEngine = create_async_engine(url, **engine_kwargs)
    _install_sqlite_connect_pragmas(self.engine)
    self._session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
        bind=self.engine, class_=AsyncSession, expire_on_commit=False
    )

`is_in_memory()` ¶

Check if this is an in-memory database.

Source code in src/chapkit/core/database.py

def is_in_memory(self) -> bool:
    """Check if this is an in-memory database."""
    return self._is_in_memory_url(self.url)

`init()` `async` ¶

Initialize database tables and configure SQLite using Alembic migrations.

Source code in src/chapkit/core/database.py

async def init(self) -> None:
    """Initialize database tables and configure SQLite using Alembic migrations."""
    # Import Base here to avoid circular import at module level
    from chapkit.core.models import Base

    # Set WAL mode first (if not in-memory)
    if not self.is_in_memory():
        async with self.engine.begin() as conn:
            await conn.exec_driver_sql("PRAGMA journal_mode=WAL;")

    # For in-memory databases or when migrations are disabled, use direct table creation
    if self.is_in_memory() or not self.auto_migrate:
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
    else:
        # For file-based databases, use Alembic migrations
        await super().init()

`SqliteDatabaseBuilder` ¶

Builder for SQLite database configuration with fluent API.

Source code in src/chapkit/core/database.py

class SqliteDatabaseBuilder:
    """Builder for SQLite database configuration with fluent API."""

    def __init__(self) -> None:
        """Initialize builder with default values."""
        self._url: str = ""
        self._echo: bool = False
        self._alembic_dir: Path | None = None
        self._auto_migrate: bool = True
        self._pool_size: int = 5
        self._max_overflow: int = 10
        self._pool_recycle: int = 3600
        self._pool_pre_ping: bool = True

    @classmethod
    def in_memory(cls) -> Self:
        """Create an in-memory SQLite database configuration."""
        builder = cls()
        builder._url = "sqlite+aiosqlite:///:memory:"
        return builder

    @classmethod
    def from_file(cls, path: str | Path) -> Self:
        """Create a file-based SQLite database configuration."""
        builder = cls()
        if isinstance(path, Path):
            path = str(path)
        builder._url = f"sqlite+aiosqlite:///{path}"
        return builder

    def with_echo(self, enabled: bool = True) -> Self:
        """Enable SQL query logging."""
        self._echo = enabled
        return self

    def with_migrations(self, enabled: bool = True, alembic_dir: Path | None = None) -> Self:
        """Configure migration behavior."""
        self._auto_migrate = enabled
        self._alembic_dir = alembic_dir
        return self

    def with_pool(
        self,
        size: int = 5,
        max_overflow: int = 10,
        recycle: int = 3600,
        pre_ping: bool = True,
    ) -> Self:
        """Configure connection pool settings."""
        self._pool_size = size
        self._max_overflow = max_overflow
        self._pool_recycle = recycle
        self._pool_pre_ping = pre_ping
        return self

    def build(self) -> SqliteDatabase:
        """Build and return configured SqliteDatabase instance."""
        if not self._url:
            raise ValueError("Database URL not configured. Use .in_memory() or .from_file()")

        return SqliteDatabase(
            url=self._url,
            echo=self._echo,
            alembic_dir=self._alembic_dir,
            auto_migrate=self._auto_migrate,
            pool_size=self._pool_size,
            max_overflow=self._max_overflow,
            pool_recycle=self._pool_recycle,
            pool_pre_ping=self._pool_pre_ping,
        )

`init()` ¶

Initialize builder with default values.

Source code in src/chapkit/core/database.py

def __init__(self) -> None:
    """Initialize builder with default values."""
    self._url: str = ""
    self._echo: bool = False
    self._alembic_dir: Path | None = None
    self._auto_migrate: bool = True
    self._pool_size: int = 5
    self._max_overflow: int = 10
    self._pool_recycle: int = 3600
    self._pool_pre_ping: bool = True

`in_memory()` `classmethod` ¶

Create an in-memory SQLite database configuration.

Source code in src/chapkit/core/database.py

@classmethod
def in_memory(cls) -> Self:
    """Create an in-memory SQLite database configuration."""
    builder = cls()
    builder._url = "sqlite+aiosqlite:///:memory:"
    return builder

`from_file(path)` `classmethod` ¶

Create a file-based SQLite database configuration.

Source code in src/chapkit/core/database.py

@classmethod
def from_file(cls, path: str | Path) -> Self:
    """Create a file-based SQLite database configuration."""
    builder = cls()
    if isinstance(path, Path):
        path = str(path)
    builder._url = f"sqlite+aiosqlite:///{path}"
    return builder

`with_echo(enabled=True)` ¶

Enable SQL query logging.

Source code in src/chapkit/core/database.py

def with_echo(self, enabled: bool = True) -> Self:
    """Enable SQL query logging."""
    self._echo = enabled
    return self

`with_migrations(enabled=True, alembic_dir=None)` ¶

Configure migration behavior.

Source code in src/chapkit/core/database.py

def with_migrations(self, enabled: bool = True, alembic_dir: Path | None = None) -> Self:
    """Configure migration behavior."""
    self._auto_migrate = enabled
    self._alembic_dir = alembic_dir
    return self

`with_pool(size=5, max_overflow=10, recycle=3600, pre_ping=True)` ¶

Configure connection pool settings.

Source code in src/chapkit/core/database.py

def with_pool(
    self,
    size: int = 5,
    max_overflow: int = 10,
    recycle: int = 3600,
    pre_ping: bool = True,
) -> Self:
    """Configure connection pool settings."""
    self._pool_size = size
    self._max_overflow = max_overflow
    self._pool_recycle = recycle
    self._pool_pre_ping = pre_ping
    return self

`build()` ¶

Build and return configured SqliteDatabase instance.

Source code in src/chapkit/core/database.py

def build(self) -> SqliteDatabase:
    """Build and return configured SqliteDatabase instance."""
    if not self._url:
        raise ValueError("Database URL not configured. Use .in_memory() or .from_file()")

    return SqliteDatabase(
        url=self._url,
        echo=self._echo,
        alembic_dir=self._alembic_dir,
        auto_migrate=self._auto_migrate,
        pool_size=self._pool_size,
        max_overflow=self._max_overflow,
        pool_recycle=self._pool_recycle,
        pool_pre_ping=self._pool_pre_ping,
    )

Models¶

`models` ¶

Base ORM classes for SQLAlchemy models.

`Base` ¶

Bases: AsyncAttrs, DeclarativeBase

Root declarative base with async support.

Source code in src/chapkit/core/models.py

class Base(AsyncAttrs, DeclarativeBase):
    """Root declarative base with async support."""

`Entity` ¶

Bases: Base

Optional base with common columns for your models.

Source code in src/chapkit/core/models.py

class Entity(Base):
    """Optional base with common columns for your models."""

    __abstract__ = True

    id: Mapped[ULID] = mapped_column(ULIDType, primary_key=True, default=ULID)
    created_at: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime.datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

Repository¶

`repository` ¶

Base repository classes for data access layer.