krossdb

An enterprise-grade, multi-database Python library that provides a unified interface and repository pattern over relational (PostgreSQL/MySQL/SQLite), document (MongoDB), and vector (Qdrant) databases.

The Python package you import is krossdb.

Why

Most projects that touch more than one kind of database end up with per-backend, ad-hoc data access code, inconsistent error handling, and no shared story for retries, caching, multi-tenancy, or observability. krossdb gives every backend the same repository/Unit-of-Work surface, the same exception hierarchy, and the same resilience and observability behavior, so application code depends on krossdb's abstractions instead of a specific driver.

Architecture at a glance

src/krossdb/
├── core/            settings, enums, the exception hierarchy, the driver
│                     exception mapper, structlog config, correlation/tenant
│                     context (contextvars)
├── interfaces/       ABCs: AbstractDatabaseAdapter, AbstractRepository,
│                     AbstractUnitOfWork, AbstractCache, domain events/outbox
├── adapters/         concrete backends, one package per DB family
│   ├── relational/    PostgreSQL / MySQL / SQLite via async SQLAlchemy,
│   │                  with master/replica read-write splitting
│   ├── document/      MongoDB via Motor
│   └── vector/        Qdrant
├── repositories/     backend-specific AbstractRepository implementations
├── uow/               Unit of Work implementations
├── cache/             Redis-backed cache-aside implementation
├── resilience/        circuit breaker + exponential-backoff retry
├── events/             in-process dispatcher + transactional outbox
├── observability/     metrics, tracing, audit trail
├── multitenancy/      tenant isolation helpers
├── factory/            DatabaseFactory: builds/owns every configured adapter
├── models/             shared Pydantic base models (Entity, AuditRecord)
├── migrations/         optional Alembic env.py wiring (see `migrations` extra)
└── cli/                `krossdb` console script (init, health, schema, ...)

Every domain entity is a Pydantic v2 model; every mutating repository method supports an idempotency key; every raised error is a KrossDBError subclass (driver exceptions are translated centrally in core/exception_mapper.py, never leaked to callers).

Installation

pip install krossdb             # from PyPI
pip install krossdb[postgres]   # + asyncpg
pip install krossdb[all]        # everything

For local development (editable install from a clone):

pip install -e ".[dev]"          # core + dev/test tooling
pip install -e ".[postgres]"     # + asyncpg
pip install -e ".[all]"          # everything

Quickstart

from krossdb import ConnectionConfig, DatabaseFactory, DatabaseKind, Entity, NodeConfig, RoutingRole
from krossdb.repositories.relational_repository import SQLAlchemyRepository

class Widget(Entity):
    name: str
    quantity: int = 0

config = ConnectionConfig(
    name="primary",
    kind=DatabaseKind.SQLITE,
    nodes=[NodeConfig(dsn="sqlite+aiosqlite:///:memory:", role=RoutingRole.MASTER)],
)

async with DatabaseFactory([config]) as factory:
    repo = SQLAlchemyRepository(factory.get("primary"), widgets_table, Widget)
    widget = await repo.create(Widget(name="gizmo", quantity=10))

See examples/quickstart.py for a complete, runnable version (table creation, CRUD, exception handling, and a transactional Unit of Work) — python examples/quickstart.py after pip install -e ".[sqlite]".

Declarative schema

Tables can also be described in YAML/JSON instead of SQLAlchemy Core code:

# schema.yaml
tables:
  - name: widgets
    columns:
      - name: id
        type: uuid
        primary_key: true
      - name: name
        type: string
        required: true
        unique: true
      - name: quantity
        type: integer
        required: true

krossdb schema create --connection primary --schema schema.yaml --config krossdb.yaml

krossdb.schema.build_metadata turns this into a real sqlalchemy.MetaData, so the same file produces correct DDL on Postgres, MySQL, and SQLite. See examples/schema.yaml for a complete example with a foreign key and an index; --metadata module.path:attribute_name still works as before for code-defined MetaData objects — pass exactly one of --schema or --metadata.

CLI

krossdb init                     # scaffold a starter krossdb.yaml
krossdb health                   # connect to every configured database and report status
krossdb schema create            # metadata.create_all() against a configured connection
krossdb schema init-migrations   # scaffold an Alembic project wired to krossdb.yaml
                                  # (requires `pip install krossdb[migrations]`)

schema init-migrations scaffolds alembic.ini + alembic/ pre-wired to read the database URL and target MetaData from krossdb.yaml instead of a hardcoded sqlalchemy.url; day-to-day migration work still goes through Alembic's own CLI (alembic revision --autogenerate, alembic upgrade head, ...). See krossdb schema init-migrations --help and krossdb.migrations.configure for details.

Versioned schema migrations

The optional migration layer compares a declarative schema with an existing database, previews additive changes and runs project-owned numbered revisions:

pip install "krossdb[migrations,sqlite]"
krossdb schema plan --connection primary --schema schema.yaml
krossdb schema migrate --connection primary --manifest migrations.yaml --dry-run
krossdb schema migrate --connection primary --manifest migrations.yaml

PostgreSQL and SQLite apply pending changes and checksum-verified history in one locked transaction. Renames, type changes and removals require explicit migrations. MySQL continues to use Alembic because its DDL implicitly commits. See the migration guide for guarantees, dialect boundaries and async usage, and a generic manifest.

Testing

pytest -m unit                   # fast, no external services required
pytest -m integration            # spins up real Postgres/MongoDB/Qdrant
                                  # containers via Testcontainers (needs Docker)
pytest                           # everything

Integration tests are automatically skipped when Docker isn't reachable, so pytest -m unit (or plain pytest in a Docker-less environment) never fails because of them.

License

MIT — see LICENSE.