ildev-mongodb 0.1.1

SOLID, extensible MongoDB client for Python using Motor (AsyncCollectionProtocol + RepositoryProtocol TCreate/TUpdate/TOut)

ildev-mongodb

SOLID, extensible MongoDB client for Python using Motor. Dict-based AsyncCollectionProtocol and typed RepositoryProtocol (TCreate / TUpdate / TOut + to_doc_create, to_doc_update, from_doc_out). Supports append-only transactional repositories, base models, aggregation, localization and JSON logging. Dependency Inversion and DI.

Requires: Python ≥3.10, Motor.

Install

pip install ildev-mongodb

Layers

1. AsyncCollectionProtocol – base protocol, document: dict[str, Any]. Full CRUD + aggregate.
2. RepositoryProtocol[TCreate, TUpdate, TOut] – typed CRUD + explicit transforms: to_doc_create, to_doc_update, from_doc_out + aggregate -> list[TOut].
3. TransactionalRepositoryProtocol[TCreate, TOut] – append-only (insert + read + aggregate), no update/delete.
4. BaseAsyncCollection – implements AsyncCollectionProtocol (dict-based).
5. BaseTypedRepository – implements RepositoryProtocol; wraps AsyncCollectionProtocol and uses the three transforms.
6. TransactionalTypedRepository – implements TransactionalRepositoryProtocol on top of BaseAsyncCollection.

Quick start (dict-based)

import asyncio
from ildev_mongodb import create_async_client

async def main():
    async with create_async_client("mongodb://localhost:27017/") as client:
        db = client.get_database("mydb")
        coll = db.get_collection("items")
        await coll.insert_one({"name": "test"})
        doc = await coll.find_one({"name": "test"})
        print(doc)

asyncio.run(main())

Quick start (typed TCreate/TUpdate/TOut)

import asyncio
from ildev_mongodb import create_async_client, BaseTypedRepository

class Item:
    def __init__(self, name: str, value: int): ...
    def to_doc(self): ...
    @classmethod
    def from_doc(cls, d: dict): ...

async def main():
    async with create_async_client("mongodb://localhost:27017/") as client:
        raw = client.get_database("mydb").get_collection("items")
        items = BaseTypedRepository(
            raw,
            to_doc_create=lambda x: x.to_doc(),
            to_doc_update=lambda u: u if isinstance(u, dict) else u.to_doc(),
            from_doc_out=Item.from_doc,
        )
        await items.insert_one(Item("x", 1))
        one = await items.find_one({"name": "x"})
        await items.update_one({"name": "x"}, {"$set": {"value": 2}})
        async for doc in items.find():
            print(doc.name, doc.value)

asyncio.run(main())

Quick start (transactional / append-only repository)

Use TransactionalTypedRepository when you only want insert + read + aggregate (event-sourcing / append-only):

import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone
from ildev_mongodb import (
    create_async_client,
    TransactionalTypedRepository,
    TransactionalCreateBase,
    TransactionalOutBase,
)


@dataclass
class EventCreate(TransactionalCreateBase):
    name: str = ""


@dataclass
class EventOut(TransactionalOutBase):
    name: str = ""


def to_doc_create(e: EventCreate) -> dict[str, object]:
    return {"name": e.name, "created_at": e.created_at}


def from_doc_out(d: dict[str, object]) -> EventOut:
    return EventOut(name=str(d["name"]))


async def main() -> None:
    async with create_async_client("mongodb://localhost:27017/") as client:
        raw = client.get_database("mydb").get_collection("events")
        events = TransactionalTypedRepository[EventCreate, EventOut](
            raw,
            to_doc_create=to_doc_create,
            from_doc_out=from_doc_out,
        )
        await events.insert_one(EventCreate(name="OrderCreated"))
        one = await events.find_one()
        print(one.name)


asyncio.run(main())

Full CRUD

AsyncCollectionProtocol (dict-based): insert_one, insert_many, find_one, find, update_one, update_many, delete_one, delete_many, count_documents.

RepositoryProtocol[TCreate, TUpdate, TOut]: same CRUD with typed payloads; exposes to_doc_create, to_doc_update, from_doc_out and .collection.

BaseTypedRepository implements RepositoryProtocol by wrapping AsyncCollectionProtocol (e.g. BaseAsyncCollection) and the three transform callables.

Base models

To standardize common fields, you can subclass these dataclasses from ildev_mongodb.models:

• TypedCreateBase – id: UUID (auto-generated) + created_at: datetime (UTC).
• TypedOutBase – id: UUID | None, created_at: datetime, updated_at: datetime.
• TypedUpdateBase – updated_at: datetime.
• TransactionalCreateBase – same as TypedCreateBase, for transactional repos.
• TransactionalOutBase – id: UUID | None, created_at: datetime (no updated_at).

Example:

from dataclasses import dataclass
from ildev_mongodb import TypedCreateBase, TypedOutBase


@dataclass
class ItemCreate(TypedCreateBase):
    name: str = ""


@dataclass
class ItemOut(TypedOutBase):
    name: str = ""

The repository type parameters must be subclasses of these base models (TCreate bound to TypedCreateBase, etc.).

Aggregation

All layers support aggregation:

• AsyncCollectionProtocol.aggregate(pipeline) -> list[dict[str, Any]]
• BaseAsyncCollection.aggregate(...) – wraps Motor’s aggregate and excludes _id from results.
• RepositoryProtocol.aggregate(pipeline) -> list[TOut]
• BaseTypedRepository.aggregate(...) / TransactionalTypedRepository.aggregate(...) – map each doc via from_doc_out.

Example:

results = await items.aggregate(
    [
        {"$match": {"name": "A"}},
        {"$project": {"_id": 0, "name": 1, "value": 1}},
    ]
)

Localization

All public exception messages can be localized via a simple JSON file:

{
  "operation_failed": "Operace selhala.",
  "connection_failed": "Připojení k databázi selhalo.",
  "unsupported_operation": "Operace není podporována."
}

Configure once at startup:

from ildev_mongodb import configure_localization

configure_localization("config/ildev_mongodb.locale.json")

Exceptions that honor localization:

• IldevMongoDBError (operation_failed)
• IldevMongoDBConnectionError (connection_failed)
• UnsupportedOperationError (unsupported_operation)

Passing an explicit message= to these constructors bypasses localization.

JSON logging

The library logs errors via the standard logging module under the logger name ildev_mongodb.

Configure JSON logging destination once at startup:

from ildev_mongodb import configure_logging

# Write JSON lines to a file
configure_logging("logs/ildev_mongodb.log", level="INFO")

# Or, to stderr by default
configure_logging()

Each log is a single JSON line, e.g.:

{"timestamp": "...", "level": "ERROR", "logger": "ildev_mongodb", "message": "database_operation_failed", "event": "db_error", "operation": "insert_one", "exc_type": "SomeMotorError", "error_class": "IldevMongoDBError"}

The library never logs URIs, full documents, or credentials; logging focuses on safe, structured error context only.

Direct client construction

from ildev_mongodb import BaseAsyncClient

async def main():
    client = BaseAsyncClient(uri="mongodb://localhost:27017/")
    try:
        db = client.get_database("mydb")
        coll = db.get_collection("items")
        # ...
    finally:
        await client.close()

License

MIT