PersistKat

Database abstraction layer for the KatCity framework. Provides async-first SQLAlchemy 2.0+ integration with repository patterns, Pydantic schema generation, and comprehensive type safety.

Features

• Repository Pattern: Generic BaseRepository with fluent query interface
• Async-First: Built on SQLAlchemy 2.0+ async engine and sessions
• Type-Safe: Full type hints with generic repository support
• Pydantic Integration: Auto-generate schemas from SQLAlchemy models
• Mixins: UUID/Int primary keys, timestamp tracking, common patterns
• Bulk Operations: Efficient bulk inserts with conflict resolution
• Migration Support: Goose-based database migrations via CLI
• Configuration: YAML-based config with environment overrides

Installation

# Core package
uv pip install persistkat

# With PostgreSQL support
uv pip install "persistkat[pg]"

# With CLI tools
uv pip install "persistkat[cli]"

# Everything
uv pip install "persistkat[all]"

Quick Start

1. Initialize Database

# SQLite (default) - creates tables directly
persistkat db init

# Or with migrations (requires goose)
persistkat db up

# Check database health
persistkat db health

Note: db init only creates tables, it never drops them. For a fresh start, manually delete the database file or use migrations with db down.

2. Define Your Models

from persistkat import Base, UUIDPKMixin, UpdateMixin
from sqlalchemy.orm import Mapped, mapped_column

class User(Base, UUIDPKMixin, UpdateMixin):
    """User model with UUID PK and timestamps."""
    
    __tablename__ = "users"
    
    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)

3. Create a Repository

from persistkat import BaseRepository
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)
    
    async def get_active_users(self) -> list[User]:
        query = self.where(User.is_active == True).order_by(User.name)
        return await self.all(query=query)

4. Use in Your Application

import asyncio
from persistkat import PersistEngine
from persistkat.config import config

async def main():
    cfg = config("config.yaml")
    engine = PersistEngine(cfg.schema.database)
    db_engine, session_factory = engine.session()
    
    async with session_factory() as session:
        repo = UserRepository(session)
        
        # Create
        user = User(name="Alice", email="alice@example.com")
        await repo.save(user)
        await session.commit()
        
        # Query
        found = await repo.find_by(email="alice@example.com")
        active = await repo.get_active_users()

CLI Tools

# Generate Pydantic schemas from models
persistkat codegen generate-schemas your_app.models --output schemas.py

# Database migrations with Goose (supports PostgreSQL, MySQL, SQLite, MSSQL, Redshift)
persistkat db create add_users_table
persistkat db up
persistkat db status
persistkat db down
persistkat db health

# Version info
persistkat version

Migration Support

PersistKat uses Goose for SQL-based database migrations. Goose supports:

• ✅ PostgreSQL - Full support including triggers, functions, and advanced features
• ✅ MySQL - Stored procedures, triggers, views
• ✅ SQLite - Lightweight migrations for development/testing
• ✅ MSSQL - Microsoft SQL Server
• ✅ Redshift - Amazon Redshift data warehouse

For other databases (Oracle, ClickHouse, CockroachDB, Snowflake, etc.):

• ✅ PersistKat engine works with ANY SQLAlchemy-supported database
• ❌ Migration CLI won't work (Goose limitation)
• Workaround: Apply migrations manually using your database client or use Alembic (Python-based migrations)

Why Goose?

• Pure SQL migrations (not Python DSL)
• Simple, version-controlled .sql files
• No code generation complexity
• Easy to audit and review changes

Configuration

PersistKat supports flexible database configuration with multiple formats:

Option 1: Full DSN (Recommended)

name: my-app
database:
  dsn: postgresql+asyncpg://user:pass@localhost:5432/mydb

Option 1b: DSN with Field Overrides (Best for Secrets)

Combine DSN structure with environment variable overrides:

name: my-app
database:
  dsn: postgresql+asyncpg://defaultuser:defaultpass@localhost:5432/mydb
  user: ${DB_USER}      # Overrides user from DSN
  password: ${DB_PASSWORD}  # Overrides password from DSN

Or in code:

export PERSISTKAT_DATABASE__USER=prod_user
export PERSISTKAT_DATABASE__PASSWORD=secret_password

The final DSN will be: postgresql+asyncpg://prod_user:secret_password@localhost:5432/mydb

Option 2: Driver-Only (Auto-Detects Dialect)

name: my-app
database:
  driver: asyncpg  # Auto-detects: dialect=postgresql
  db: myapp-dev
  host: localhost
  port: 5432
  user: postgres
  password: postgres

Option 3: Individual Fields (Most Explicit)

name: my-app
database:
  dialect: postgresql
  driver: asyncpg
  db: myapp-dev
  user: postgres
  password: postgres
  host: localhost
  port: 5432
  debug: false

Supported Drivers

PostgreSQL:

• asyncpg (recommended, fastest)
• psycopg (psycopg3, also excellent)

SQLite:

• aiosqlite (recommended)

MySQL:

• aiomysql
• asyncmy

Others:

• Any async SQLAlchemy driver (aioodbc, custom drivers, etc.)

Environment Variable Overrides

export PERSISTKAT_DATABASE__DSN=postgresql+asyncpg://localhost/prod-db
export PERSISTKAT_DATABASE__DEBUG=true

Limitations & Alternatives

Migration Tool Limitations

The persistkat db commands use Goose, which supports most common databases but not all SQLAlchemy dialects.

If you need migrations for unsupported databases:

1. Use Alembic - SQLAlchemy's official migration tool (Python-based, not pure SQL)

   uv pip install alembic
   alembic init migrations
   alembic revision -m "create tables"
   alembic upgrade head
   

2. Apply SQL manually - Use your database's native client

   # PostgreSQL
   psql -d mydb -f migrations/001_init.sql
   
   # MySQL
   mysql -u root -p mydb < migrations/001_init.sql
   

3. Request Goose support - Submit feature request to Goose project

Engine vs Migration Support

Database	PersistKat Engine	Migration CLI
PostgreSQL	✅ Full support	✅ Via Goose
MySQL	✅ Full support	✅ Via Goose
SQLite	✅ Full support	✅ Via Goose
MSSQL	✅ Full support	✅ Via Goose
Redshift	✅ Full support	✅ Via Goose
Oracle	✅ Full support	⚠️ Manual SQL
ClickHouse	✅ Full support	⚠️ Manual SQL
CockroachDB	✅ Full support	⚠️ Use CockroachDB's tools
Snowflake	✅ Full support	⚠️ Use Snowflake's tools

Development

• Run tests: make test
• Skip database tests: SKIP_DB_TESTS=true make test or pytest -m "not db"
• Run only database tests: pytest -m db
• Run linter & formatter: make check and make fix
• Run type checker: make pyright

Running Tests

PersistKat tests support multiple database backends:

• SQLite (default) - No setup required, uses temporary in-memory database
• PostgreSQL - Requires running PostgreSQL server
• Both - Runs tests against both backends

Database Backend Selection:

# Use SQLite (default, no setup needed)
make test

# Use PostgreSQL (requires running database)
TEST_DB_BACKEND=postgresql make test

# Test against both backends
TEST_DB_BACKEND=both make test

# Skip database tests entirely
SKIP_DB_TESTS=true make test
# or
pytest -m "not db"

Test Categories:

1. Unit tests - Don't require a database connection
2. Database tests - Marked with @pytest.mark.db
3. SQLite-specific - Marked with @pytest.mark.sqlite
4. PostgreSQL-specific - Marked with @pytest.mark.postgresql

Run specific backend tests:

# Only SQLite tests
pytest -m "db and sqlite"

# Only PostgreSQL tests (requires database on localhost:5434)
pytest -m "db and postgresql"

Documentation

See the KatCity monorepo README for architecture and design principles.

For detailed guides:

• Repository pattern best practices
• Transaction management
• FastAPI integration
• Bulk operations
• Schema generation

License

BSD 3-Clause License - See LICENSE for details.