Toy ThingDB implementation using python and mongodb

thingdb.py

import asyncio
from abc import ABC
from enum import Enum
from typing import Optional

from odmantic import AIOEngine
from odmantic import Model
from odmantic.query import SortExpression


class TypesEnum(Enum):
    WORK = '/type/work'
    AUTHOR = '/type/author'

    DELETE = '/type/delete'


class ThingModel(Model, ABC):
    key: str
    type: str
    data: dict
    version: int


class EditTrackModel(Model, ABC):
    key: str
    from_version: int
    to_version: int
    actor: str


class Thing:

    def __init__(self, db_engine):
        self.db_engine = db_engine

    async def get(self, key: str) -> Optional[ThingModel]:
        return await self.db_engine.find_one(
            ThingModel,
            ThingModel.key == key,
            sort=ThingModel.version.desc()
        )

    async def create(self, key: str, thing_type: TypesEnum, data: dict) -> ThingModel:
        existing_thing: Optional[ThingModel] = await self.get(key)
        if existing_thing:
            raise Exception('data already exists with key')
        thing = ThingModel(
            data=data,
            version=1,
            key=key,
            type=str(thing_type)
        )
        return await self.db_engine.save(thing)

    async def update(self, key: str, data: dict) -> ThingModel:
        existing_thing: Optional[ThingModel] = await self.get(key)
        if not existing_thing:
            raise Exception('Nothing to update')
        thing = ThingModel(
            data=data,
            version=existing_thing.version + 1,
            key=existing_thing.key,
            type=existing_thing.type
        )
        return await self.db_engine.save(thing)

    async def delete(self, key: str) -> ThingModel:
        existing_thing: Optional[ThingModel] = await self.get(key)
        if not existing_thing:
            raise Exception('Nothing to delete')
        thing = ThingModel(
            data=existing_thing.data,
            version=existing_thing.version + 1,
            key=existing_thing.key,
            type=str(TypesEnum.DELETE)
        )
        return await self.db_engine.save(thing)


async def main():
    work_key = 'OL123W'
    engine = AIOEngine()
    thing_db = Thing(engine)
    await thing_db.create(
        key=work_key,
        thing_type=TypesEnum.WORK,
        data={'name': 'Harry potter'}
    )
    await thing_db.update(work_key, data={'name': 'Harry potter and the philosophers stone'})
    await thing_db.delete(work_key)
    print(await thing_db.get(work_key))


if __name__ == '__main__':
    asyncio.run(main())