NoSQLRunner

from mango.nosql_runner import NoSQLRunner

Abstract base class for all database backends. Implement this to add support for any NoSQL database.

Interface

class NoSQLRunner(ABC):

    @abstractmethod
    def connect(self, connection_string: str, **kwargs) -> None: ...

    @abstractmethod
    def execute_query(self, operation: QueryRequest) -> list[dict[str, Any]]: ...

    @abstractmethod
    def introspect_schema(self) -> dict[str, SchemaInfo]: ...

    @abstractmethod
    def get_sample_documents(self, collection: str, n: int = 5) -> list[dict]: ...

    @abstractmethod
    def list_collections(self) -> list[str]: ...

    @abstractmethod
    def get_indexes(self, collection: str) -> list[dict]: ...

execute_query() returns a list of JSON-safe row dicts: find/aggregate return the documents, count returns [{"count": N}], distinct returns [{field: value}, ...].

QueryRequest

@dataclass
class QueryRequest:
    operation: Literal["find", "aggregate", "count", "distinct"]
    collection: str
    filter: dict | None = None
    pipeline: list[dict] | None = None
    projection: dict | None = None
    sort: dict | None = None
    limit: int | None = None
    distinct_field: str | None = None

Exceptions

All inherit from MangoError.

Built-in implementation

Custom implementation

from typing import Any
from mango.nosql_runner import NoSQLRunner
from mango.core.types import QueryRequest, SchemaInfo

class MyRedisRunner(NoSQLRunner):

    def connect(self, connection_string: str, **kwargs) -> None:
        self._client = redis.from_url(connection_string)

    def execute_query(self, operation: QueryRequest) -> list[dict[str, Any]]:
        # Translate QueryRequest to Redis commands, return row dicts
        ...

    def list_collections(self) -> list[str]:
        # In Redis: return key patterns or namespaces
        ...

    # implement remaining methods...