Keble-DB

Lightweight database toolkit for MongoDB (PyMongo/Motor), SQL (SQLModel/SQLAlchemy), Qdrant, and Neo4j. Includes sync + async CRUD base classes, a shared QueryBase, a Db session manager, FastAPI deps (ApiDbDeps), and Redis namespace wrappers.

Neo4j sync and async drivers share typed pool-health settings. Idle connections are checked after 60 seconds by default and recycled after 900 seconds; services may override both values through their DbSettingsABC implementation.

Installation

pip install "keble-db>=1.10.1,<2"

The current source release target is 1.10.1. It requires keble-helpers >=2,<3, keeping DB schemas and wrappers on the same typed catalog/money foundation as service consumers. This release line owns the shared owner-token Redis lease, strict UTC Mongo decoding, and isolated DB test utilities consumed by the raw-data provider packages.

Core API (import from keble_db)

• Queries/types: DbSettingsABC, QueryBase, ObjectId, Uuid
• CRUD:
  • MongoCRUDBase[Model]
  • SqlCRUDBase[Model]
  • QdrantCRUDBase[Payload, Vector] (+ Record)
  • Neo4jCRUDBase[Model]
• Connections/DI: Db(settings: DbSettingsABC), ApiDbDeps(db)
• Redis: ExtendedRedis, ExtendedAsyncRedis, RedisLeaseManager, RedisLease
• Mongo helpers: build_mongo_find_query, merge_mongo_and_queries, merge_mongo_or_queries
• BSON Decimal boundary: encode_decimal128, decode_decimal128, BsonDocument

Async methods are prefixed with a (e.g. afirst, aget_multi, adelete). Db.try_close_async() uses redis-py's supported aclose() path for async Redis clients and retains the typed cleanup behavior for the other database clients. The package local-full suite drops only its exact __keble_db__pytest__ Mongo database and Qdrant collection during session teardown.

Redis owner-token leases

Use RedisLeaseManager for cross-process single-flight or short-lived command ownership. Acquisition is atomic SET NX PX; renewal and release execute Lua compare-and-mutate operations, so an expired/stale owner cannot renew or delete its successor's lease.

from redis.asyncio import Redis

from keble_db import RedisLeaseManager


redis = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
leases = RedisLeaseManager(client=redis, namespace="raw-data-api")
lease = await leases.acquire(resource="keepa:asin:B000000001", ttl_ms=30_000)
if lease is not None:
    async with lease:
        # Recheck the durable cache, fetch upstream once, then persist.
        await refresh_cache()

The namespace and resource are both part of physical key identity. Always use an expiry longer than the bounded critical section, call renew() for an intentionally extended operation, and use async with so normal failure or cancellation releases ownership promptly.

Agent Deps

AgentDbDeps owns database-wide pydantic-ai runtime dependencies and the optional outer progress_task.

1. Package-specific deps should inherit AgentDbDeps.
2. Package-specific state should live under one package namespace such as .segmenting, .positioning, or .task.
3. Shared request progress should use deps.progress_task, not nested package fields such as deps.segmenting.progress_task.

class SegmentingAgentDeps(AgentDbDeps):
    """DB deps plus segmenting-owned runtime namespace."""

    segmenting: SegmentingAgentContext

QueryBase expectations

QueryBase fields: filters, order_by, offset, limit, id, ids.

• Mongo: filters is a Mongo query dict; order_by is [(field, ASCENDING|DESCENDING)]; offset/limit are int.
• SQL: filters is a list of SQLAlchemy expressions; order_by is an expression or list; offset/limit are int.
• Qdrant:
  • search(): filters is a Qdrant filter dict, offset is int|None, limit defaults to 100.
  • scroll(): offset is PointId|None (point id) and limit is required; ordering uses order_by (str or Qdrant OrderBy) or falls back to QueryBase.order_by. Qdrant requires a payload range index for the ordered key. Example: from qdrant_client.models import PayloadSchemaType; crud.ensure_payload_indexes(client, payload_indexes={"id": PayloadSchemaType.INTEGER}).
• Neo4j: filters is a dict of property predicates (operators: $gt, $gte, $lt, $lte, $in, $contains, $startswith, $endswith); order_by is [(field, "asc"|"desc")]; offset/limit are int.

Examples

MongoDB

from pydantic import BaseModel
from pymongo import MongoClient, DESCENDING

from keble_db import MongoCRUDBase, QueryBase


class User(BaseModel):
    name: str
    age: int


crud = MongoCRUDBase(User, collection="users", database="app")
m = MongoClient(
    "mongodb://localhost:27017",
    uuidRepresentation="standard",
    tz_aware=True,
)

crud.create(m, obj_in=User(name="Alice", age=30))
users = crud.get_multi(
    m,
    query=QueryBase(filters={"age": {"$gte": 18}}, order_by=[("age", DESCENDING)]),
)

Db.get_mongo(), Db.get_amongo(), Db.aget_amongo(), and the shared pytest fixtures always construct clients with tz_aware=True. MongoDB stores datetimes at millisecond precision and returns aware UTC values through these boundaries; persisted domain models should normalize sub-millisecond precision before an exact in-memory-versus-restored equality check.

Exact Decimal documents

encode_decimal128 and decode_decimal128 are the only public recursive Decimal/BSON conversion boundary. Both accept a complete typed mapping, return a new document, recurse through nested mappings and arrays, and preserve exact Decimal precision. MongoCRUDBase uses the same functions on every modeled write and read.

from decimal import Decimal

from keble_db import decode_decimal128, encode_decimal128


encoded = encode_decimal128(
    {"usage": [{"unit_quantity": Decimal("19.125")}]}
)
decoded = decode_decimal128(encoded)
assert decoded["usage"][0]["unit_quantity"] == Decimal("19.125")

Side effects if changes:

• every Mongo CRUD consumer shares this write/read conversion seam;
• Data Infra video usage projections require nested Decimal round trips;
• downstream packages must import this API instead of redeclaring serializers.

SQL (SQLModel)

import uuid
from typing import Optional

from sqlmodel import Field, Session, SQLModel, create_engine

from keble_db import QueryBase, SqlCRUDBase


class User(SQLModel, table=True):
    id: Optional[str] = Field(
        default_factory=lambda: str(uuid.uuid4()), primary_key=True
    )
    name: str
    age: int


engine = create_engine("sqlite:///db.sqlite")
SQLModel.metadata.create_all(engine)
crud = SqlCRUDBase(User, table_name="users")

with Session(engine) as s:
    created = crud.create(s, obj_in=User(name="Alice", age=30))
    found = crud.first(s, query=QueryBase(id=created.id))

Qdrant

Requires qdrant-client>=1.16.0 (uses query_points).

from pydantic import BaseModel
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, PayloadSchemaType, VectorParams

from keble_db import QdrantCRUDBase, QueryBase


class Payload(BaseModel):
    id: int
    name: str


class Vector(BaseModel):
    vector: list[float]


client = QdrantClient(host="localhost", port=6333)
client.recreate_collection(
    collection_name="items",
    vectors_config={"vector": VectorParams(size=3, distance=Distance.COSINE)},
)

crud = QdrantCRUDBase(Payload, Vector, collection="items")
crud.ensure_payload_indexes(
    client,
    payload_indexes={"id": PayloadSchemaType.INTEGER},
)
crud.create(client, Vector(vector=[0.1, 0.2, 0.3]), Payload(id=1, name="a"), "p1")
hits = crud.search(
    client,
    vector=[0.1, 0.2, 0.3],
    vector_key="vector",
    query=QueryBase(filters={"must": [{"key": "id", "match": {"value": 1}}]}, limit=5),
)

If you have per-embedder collections (common in RAG), use deterministic naming:

collection = QdrantCRUDBase.derive_collection_name(
    base="items",
    embedder_id="text-embedding-3-small",
)
crud = QdrantCRUDBase(Payload, Vector, collection=collection)

Neo4j

from pydantic import BaseModel
from neo4j import GraphDatabase

from keble_db import Neo4jCRUDBase, QueryBase


class Person(BaseModel):
    id: int
    name: str


driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password"))
crud = Neo4jCRUDBase(Person, label="Person", id_field="id")

with driver.session() as s:
    crud.create(s, obj_in=Person(id=1, name="Alice"))
    people = crud.get_multi(s, query=QueryBase(filters={"id": 1}))

Db + FastAPI

Db(settings) builds clients from a DbSettingsABC implementation (see keble_db/schemas.py; package-local tests keep sample settings in keble_db/testing/local_config.py).

SQL Pool Limits

Db uses bounded SQLAlchemy pools for every sync and async read/write engine. The defaults are intentionally conservative because each backend process owns separate read and write engines:

• sync pool size: 3
• sync max overflow: 2
• async pool size: 3
• async max overflow: 2
• pool recycle: 1800 seconds
• pool timeout: 30 seconds

Override these properties on your DbSettingsABC implementation when a service has a larger PostgreSQL connection budget:

class Settings(DbSettingsABC):
    SQL_ASYNC_POOL_SIZE: int = 4

    @property
    def sql_async_pool_size(self) -> int:
        """Return the async SQL pool size owned by one process."""
        return self.SQL_ASYNC_POOL_SIZE

The package validates pool settings at startup so invalid values fail before traffic can exhaust PostgreSQL with oversized connection pools. ApiDbDeps(db) exposes FastAPI-friendly generator dependencies such as get_mongo, get_amongo, get_read_sql, get_write_asql, get_qdrant, get_neo4j_session, plus Redis equivalents. Neo4j dependency behavior:

• get_neo4j_session and get_async_neo4j_session yield session objects.
• get_aneo4j yields an AsyncDriver.

More runnable examples

See tests/integration/crud/ and tests/unit/test_api_deps.py.

Testing

keble-db owns the canonical database testing helpers for Keble Python repos. Use keble_db.testing before creating ad hoc DB clients or cleanup logic.

Shared helpers include:

• create_test_namespace(...) for one namespace across Postgres, Mongo, Qdrant, Neo4j, and Redis.
• qdrant_memory_client() / async_qdrant_memory_client() for fast local-lite vector tests.
• eventually(...) / eventually_sync(...) for predicate-based polling instead of random sleeps.
• keble_db.testing.pytest_plugin for canonical markers and opt-in DB fixtures.
• Mongo fixtures decode BSON datetimes as timezone-aware UTC and drop only their process-unique database after each test.

Executable tests must live under canonical layer-first folders:

tests/unit/
tests/contract/
tests/integration/
tests/live/
tests/evals/
tests/db_stack/

Do not add new flat tests/test_*.py, tests/mock/, tests/irl/, or provider-named layer folders. DB behavior belongs in tests/integration/ or tests/db_stack/ with canonical markers and isolated namespaces.

Local-full default command:

RUN_INTEGRATION=1 RUN_REAL_DB=1 RUN_LOCAL_STACK=1 RUN_DB_STACK=1 \
  uv run pytest -q -m "not live and not container"

The local-full command is the normal Keble development proof. Qdrant tests use local :memory: mode unless they explicitly test server-only behavior such as payload indexes. Tests that require real Postgres, Mongo, Redis, Neo4j, or server-backed Qdrant carry local_stack and now run in the local-full command with isolated namespaces and cleanup.

Portable-offline fallback:

uv run pytest -q -m "not live and not slow and not eval and not local_stack and not db_stack and not container"

Focused dependency tests:

RUN_INTEGRATION=1 RUN_REAL_DB=1 RUN_LOCAL_STACK=1 uv run pytest -q -m integration
RUN_DB_STACK=1 RUN_LOCAL_STACK=1 uv run pytest -q -m db_stack
uv run pytest -q -m eval

Selecting -m eval is the eval opt-in. There is no generic RUN_EVALS gate.

Useful environment variables for integration fixtures:

• POSTGRES_TEST_DSN
• POSTGRES_ASYNC_TEST_DSN
• MONGO_TEST_URI
• REDIS_URI
• QDRANT_HOST
• QDRANT_PORT
• NEO4J_TEST_URI
• NEO4J_TEST_USER
• NEO4J_TEST_PASSWORD
• NEO4J_TEST_DATABASE

If these test-specific variables are not set, keble_db.testing looks for the umbrella keble.backend/.env file and maps backend names such as MONGO_DB_URI, POSTGRES_*, REDIS_URI, QDRANT_*, and NEO4J_* into the shared fixtures. Set KEBLE_BACKEND_ENV_FILE=/path/to/.env when running from a worktree or CI location where the backend env file is not a sibling of the umbrella root. Explicit process environment variables always win over backend dotenv values.

For one-off local service overrides, copy tests/assets/config.example.json to the ignored tests/assets/config.json and edit the copy. Do not commit real DB credentials or developer-local service passwords; the committed example uses placeholder values only.

Run pyright from this package root after Python changes:

npx --yes pyright .