keble-db 1.8.0

Keble db

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.

Installation

pip install keble-db

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
• Mongo helpers: build_mongo_find_query, merge_mongo_and_queries, merge_mongo_or_queries

Async methods are prefixed with a (e.g. afirst, aget_multi, adelete).

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")

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)]),
)

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 or tests/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/test_crud/ and tests/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.

Default fast command:

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

The default command must stay offline and fast. Qdrant tests use local :memory: mode unless they explicitly test server-only behavior such as payload indexes.

Real dependency tests are opt-in:

RUN_INTEGRATION=1 uv run pytest -m integration
RUN_DB_STACK=1 uv run pytest -m db_stack

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

Run pyright from this package root after Python changes:

npx --yes pyright .