lexigram-nosql

NoSQL document store support for the Lexigram Framework (MongoDB, DynamoDB, Firestore).

Overview

lexigram-nosql provides async document-store backends behind a clean protocol interface. It ships with a MongoDB driver (Motor-based), a fluent query builder, aggregation pipelines, the repository pattern with specifications, a migration manager, and Named DI multi-backend support. The MongoDB driver is registered through the container; DynamoDB and Firestore backends are available as direct-use classes (lexigram.nosql.backends.dynamodb, lexigram.nosql.backends.firestore).

Full documentation: docs.lexigram.dev

Install

uv add lexigram lexigram-nosql

# With MongoDB support
uv add "lexigram-nosql[mongodb]"
# With DynamoDB support
uv add "lexigram-nosql[dynamodb]"
# With Firestore support
uv add "lexigram-nosql[firestore]"

Quick Start

from lexigram import Application
from lexigram.di.module import Module, module
from lexigram.nosql import NoSQLModule
from lexigram.nosql.config import MongoDBConfig, NoSQLConfig
from lexigram.contracts.data.nosql.nosql import DocumentStoreProtocol


@module(
    imports=[
        NoSQLModule.configure(
            NoSQLConfig(
                driver="mongodb",
                mongodb=MongoDBConfig(
                    uri="mongodb://localhost:27017",
                    database="myapp",
                ),
            )
        )
    ]
)
class AppModule(Module):
    pass


async def main() -> None:
    async with Application.boot(modules=[AppModule]) as app:
        store = await app.container.resolve(DocumentStoreProtocol)
        collection = store.collection("users")

        await collection.insert_one({"name": "Alice", "age": 30})
        async for user in collection.find({"age": {"$gte": 25}}):
            print(user)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Configuration

Zero-config usage: Call NoSQLModule.configure() with no arguments to use all defaults.

Option 1 — YAML file

# application.yaml
nosql:
  driver: "mongodb"
  mongodb:
    uri: "mongodb://localhost:27017"
    database: "myapp"
    max_pool_size: 100

Option 2 — Profiles + Environment Variables (recommended)

export LEX_NOSQL__DRIVER=mongodb
export LEX_NOSQL__MONGODB__URI=mongodb://localhost:27017
export LEX_NOSQL__MONGODB__DATABASE=myapp

Option 3 — Python

from lexigram.nosql import NoSQLModule
from lexigram.nosql.config import NoSQLConfig, MongoDBConfig

NoSQLModule.configure(
    NoSQLConfig(
        driver="mongodb",
        mongodb=MongoDBConfig(uri="mongodb://localhost:27017", database="myapp"),
    )
)

Config reference

NoSQLConfig

MongoDBConfig

Module Factory Methods

Key Features

• MongoDB backend — async Motor-based with connection pooling and retry logic
• Query builder — type-safe fluent API for MongoDB queries and projections
• Aggregation pipelines — composable pipeline stages for complex aggregations
• Repositories — base DocumentRepository pattern with specification support
• Migration manager — index creation, field operations, and collection management
• Named DI multi-backend — multiple backends registered via Annotated[DocumentStoreProtocol, Named("analytics")]
• Session and transaction context managers — mongodb_session() and mongodb_transaction() (lexigram.nosql.backends.mongodb.session) for ACID operations

Testing

lexigram-nosql ships no in-memory backend — stub() uses the MongoDB driver, so boot requires a reachable MongoDB. Point it at a test instance:

from lexigram.nosql.config import MongoDBConfig, NoSQLConfig

config = NoSQLConfig(
    driver="mongodb",
    mongodb=MongoDBConfig(uri="mongodb://localhost:27017", database="testdb"),
)

async with Application.boot(modules=[NoSQLModule.stub(config)]) as app:
    store = await app.container.resolve(DocumentStoreProtocol)
    collection = store.collection("users")
    await collection.insert_one({"name": "Alice"})  # requires a live test MongoDB

Key Source Files