ildev-mongodb 0.1.0

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 RepositoryProtocol (TCreate/TUpdate/TOut + to_doc_create, to_doc_update, from_doc_out). 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.
2. RepositoryProtocol[TCreate, TUpdate, TOut] – typed CRUD + explicit transforms: to_doc_create, to_doc_update, from_doc_out.
3. BaseAsyncCollection – implements AsyncCollectionProtocol (dict-based).
4. BaseTypedRepository – implements RepositoryProtocol; wraps AsyncCollectionProtocol and uses the three transforms.

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

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.

Design (SOLID, DI)

• AsyncCollectionProtocol – dict-based collection; BaseAsyncCollection implements it.
• RepositoryProtocol – typed repository with explicit doc transforms; BaseTypedRepository implements it.
• create_async_client() – DI entry point; yields BaseAsyncClient.

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

PyPI

pip install build twine
python -m build
twine upload dist/*

License

MIT