43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
import json
|
|
from datetime import datetime
|
|
from typing import Optional, Any, Union
|
|
from uuid import UUID
|
|
|
|
#
|
|
# def uuid_encoder():
|
|
# print(obj)
|
|
# if isinstance(obj, UUID):
|
|
# # if the obj is uuid, we simply return the value of uuid
|
|
# return obj.hex
|
|
# return json.JSONEncoder.default(self, obj)
|
|
#
|
|
import jsonpatch
|
|
from pydantic import Field
|
|
|
|
from core.models.base import BaseModel
|
|
|
|
|
|
class DataModelPatch(BaseModel):
|
|
id: Optional[Union[UUID, str]] = Field(alias='_id')
|
|
patch: Optional[Any]
|
|
ref_id: UUID
|
|
collection: str
|
|
owner: UUID
|
|
created_at: Optional[datetime]
|
|
|
|
class DB:
|
|
collection: str = 'revisions'
|
|
|
|
|
|
async def set_revision(collection: str, old: str | dict, new: str | dict, ref_id: UUID, user_id: UUID):
|
|
old = json.dumps(old, default=str) if isinstance(old, dict) else old
|
|
new = json.dumps(new, default=str) if isinstance(new, dict) else new
|
|
patch = jsonpatch.JsonPatch.from_diff(
|
|
json.loads(new), json.loads(old)
|
|
)
|
|
await DataModelPatch(
|
|
patch=json.loads(patch.to_string()),
|
|
ref_id=str(ref_id),
|
|
collection=collection,
|
|
owner=str(user_id)).init_id().c_at().save()
|