tokka 0.1.1b0

A thin async layer between Pydantic and MongoDB

A thin async layer between Pydantic and MongoDB.

[DOCS] [SOURCE]

What is

Tokka is a MongoDB/Motor Async wrapper for Pydantic models.

As a heavy Pydantic/ FastAPI user, I faced myself writing some recurrent boilerplate code to make things work and remain pleasantly readable when on projects involving MongoDB.

Nowadays, Pydantic-core is written in Rust, and it's blazing fast. So, Tokka abuses from Pydantic's model_dump method to serialize Pydantic models into Dict/MongoDB documents.

No magic, no complex things, only dump. I tried to keep the code as simple as possible and to not fall deep into Pydantic's internals. I also tried to keep the code as close as possible to Pymongo's API, so it's familiar to understand and use. But I took some liberties to make things more Pythonic adding some syntactic sugar here and there, as well as some extra agnostic functionalities. In addition, Pymongo methods has some pretty strange kwargs documentation and a not-so-good type annotations, then I worked on trying make it a little bit friendly.

Personally, I see Tokka as an ingenuous package for lazy people like me. If you can make some use of it or it make you write less code, I'll be glad.

Installation

pip install tokka

Quick usage

from pydantic import BaseModel

from tokka import Database
from tokka import Collection

import asyncio


class User(BaseModel):
    """Sample data."""

    name: str
    email: str


class DB(Database):
    """A tokka.Database subclass to easily accesst the your collections."""

    @property
    def users(self) -> Collection:
        return self.get_collection("users")

if __name__ == "__main__":
    db = DB("sampleDB", connection="YOUR MONGODB URI")
    user1 = User(name="John Doe", email="[e-mail]")
    user2 = User(name="Emma Soo", email="[e-mail]")

    async def tasks() -> None:
        insert_results = await asyncio.gather(
            db.users.insert_one(user1),
            db.users.find_one(user1, filter_by="name"),
        )

        print(insert_results)

        replace_one_results = await asyncio.gather(
            db.users.replace_one(user1, user2, filter_by="email"),
            db.users.find_one(user2, filter_by="name"),
        )

        print(replace_one_results)

        find_one_and_delete_results = await asyncio.gather(
            db.users.find_one_and_delete(user2, filter_by="name"),
        )

        print(find_one_and_delete_results)
        

    asyncio.run(tasks())
    db.close()

Docs

Tokka is almost a syntatic-sugar package wrapping Motor and Pydantic.

Docstrings and repo samples should be enought to get you running in a few seconds.

Testing

Tokka tests run using transactions. Therefore, the containerized Mongo MUST be started as a replica set.

docker run --rm -d -p 27017:27017 \
    -h $(hostname) --name mongo mongo:7.0 \
    --replSet=tokka && sleep 4

docker exec mongo mongosh --quiet --eval "rs.initiate();"

After container startup, simple run pytest:

pytest -s  

Benchmarking

Formal benchmarks are still necessary, but initial executions showed an impact of less than <0.1s using Tokka.

License

This project is licensed under the terms of the MIT license.