ildev-mongodb 0.1.1
SOLID, extensible MongoDB client for Python using Motor (AsyncCollectionProtocol + RepositoryProtocol TCreate/TUpdate/TOut)
License: MIT
Requires Python: >=3.10
Keywords: mongodb, async, asyncio, motor, database

# 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

```bash
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)

```python
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)

```python
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):

```python
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:

```python
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:

```python
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:

```json
{
  "operation_failed": "Operace selhala.",
  "connection_failed": "Připojení k databázi selhalo.",
  "unsupported_operation": "Operace není podporována."
}
```

Configure once at startup:

```python
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:

```python
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.:

```json
{"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

```python
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