krossdb gives PostgreSQL, MySQL, SQLite, MongoDB, and Qdrant the same repository & Unit-of-Work surface, the same exception hierarchy, and the same retries, caching, and observability — so your application code depends on an abstraction, not a driver.
Most projects that touch more than one kind of database end up with per-backend, copy-pasted data access code, inconsistent error handling, and no shared story for retries, caching, multi-tenancy, or observability.
KrossDBError hierarchyTenantScopedRepository makes isolation the default, not an afterthoughtEvery domain entity is a Pydantic v2 model. Every mutating repository method supports
an idempotency key. Every raised error is a KrossDBError
subclass — driver exceptions never leak past the adapter boundary.
src/krossdb/ ├── core/ settings, exception hierarchy, driver-error mapper, │ structlog config, correlation/tenant context ├── interfaces/ AbstractDatabaseAdapter, AbstractRepository, │ AbstractUnitOfWork, AbstractCache, events/outbox ├── adapters/ one package per backend family │ ├── relational/ Postgres / MySQL / SQLite via async SQLAlchemy, │ │ read-write splitting across replicas │ ├── 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 (`migrations` extra) └── cli/ `krossdb` console script (init, health, schema create/init-migrations)
The same CRUD and query surface across every backend, with atomic multi-write transactions via SQLAlchemyUnitOfWork.
Partial bulk-insert success via savepoints, with task-local session binding that's safe under concurrent requests.
Exponential-backoff retry and a circuit breaker sit in front of every adapter call, tuned per connection.
TenantScopedRepository injects tenant filtering automatically, driven by contextvar-based tenant context.
A drop-in cache-aside layer plus a generic idempotency-key manager for safe write retries.
An in-process event dispatcher and a transactional outbox — in-memory or SQL table-backed — for reliable side effects.
Metrics and tracing helpers out of the box, plus an audit trail recorder for every mutation.
structlog with correlation IDs and automatic PII/secret masking, threaded through contextvars.
Weighted load balancing across read replicas for the relational adapters, configured per ConnectionConfig.
krossdb schema init-migrations scaffolds a starter Alembic project that resolves its DB URL and target metadata from your own krossdb.yaml — no hardcoded sqlalchemy.url.
No external services required — this runs against an in-memory SQLite adapter. The
full runnable version lives at examples/quickstart.py.
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) # basic CRUD widget = await repo.create(Widget(name="gizmo", quantity=10)) updated = await repo.update(widget.id, {"quantity": 25}) # an atomic transaction spanning multiple writes async with SQLAlchemyUnitOfWork(factory.get("primary")) as uow: await repo.create(Widget(name="widget-a", quantity=1)) await repo.create(Widget(name="widget-b", quantity=2)) await uow.commit()
| Family | Backends | Driver | Install extra | Notable |
|---|---|---|---|---|
| Relational | PostgreSQL, MySQL, SQLite | async SQLAlchemy | [postgres] [mysql] [sqlite] |
Master/replica read-write splitting, weighted load balancing |
| Document | MongoDB | Motor | [mongodb] |
Async document repository over the same interface |
| Vector | Qdrant | qdrant-client | [vector] |
Similarity search through the standard repository API |
Integration tests spin up real Postgres/MongoDB/Qdrant via Testcontainers and are
skipped automatically when Docker isn't reachable — pytest -m unit
never fails because of them.
Fast, isolated tests against mocked adapters via pytest-mock. No external services.
Real Postgres, MongoDB, and Qdrant containers via Testcontainers. Needs Docker.
Runs the full suite; integration tests skip cleanly without Docker reachable.