Skip to content

Instantly share code, notes, and snippets.

@Freakachoo

Freakachoo/gist:f41bf22f118c8117bf21d0443645cb3e

Created July 15, 2025 14:37
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Freakachoo/f41bf22f118c8117bf21d0443645cb3e to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Freakachoo/f41bf22f118c8117bf21d0443645cb3e to your computer and use it in GitHub Desktop.
Download ZIP
MongoDB with versioning
Raw
gistfile1.txt
```python
from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine
client = AsyncIOMotorClient("mongodb://localhost:27017")
engine = AIOEngine(motor_client=client, database="test_db")
def odmantic_to_pydantic(document: MyDocument) -> DocumentResponse:
return DocumentResponse(id=str(document.id), data=document.data, latest_version=document.latest_version)
async def create_document(data: Dict[str, Any]):
document = MyDocument(data=data, latest_version=1)
await engine.save(document)
return odmantic_to_pydantic(document)
async def update_document(document_id: str, new_data: Dict[str, Any]):
document = await engine.find_one(MyDocument, MyDocument.id == document_id)
if not document:
raise HTTPException(status_code=404, detail="Document not found")
diff = calculate_diff(document.data, new_data)
new_version = document.latest_version + 1
document.data = new_data
document.latest_version = new_version
await engine.save(document)
diff_entry = DocumentDiff(
document_id=str(document.id),
version=new_version,
diff=diff,
timestamp=datetime.utcnow()
)
await engine.save(diff_entry)
return odmantic_to_pydantic(document)
async def get_document_versions(document_id: str):
versions = await engine.find(DocumentDiff, DocumentDiff.document_id == document_id)
return [version.diff for version in versions]
async def get_specific_version(document_id: str, version: int) -> Dict[str, Any]:
document = await engine.find_one(MyDocument, MyDocument.id == document_id)
if not document:
raise HTTPException(status_code=404, detail="Document not found")
versions = await engine.find(DocumentDiff, DocumentDiff.document_id == document_id)
versions.sort(key=lambda x: x.version)
data = document.data.copy()
for v in versions:
if v.version > version:
break
data = apply_diff(data, v.diff)
return data
# Example usage
import asyncio
async def main():
# Create a new document
doc = await create_document({"key": "value"})
print(f"Created document: {doc}")
# Update the document
updated_doc = await update_document(str(doc.id), {"key": "new value"})
print(f"Updated document: {updated_doc}")
# Get document versions (diffs)
versions = await get_document_versions(str(doc.id))
print("Document versions (diffs):")
for version in versions:
print(version)
# Get a specific version of the document
specific_version = await get_specific_version(str(doc.id), 1)
print(f"Specific version (1): {specific_version}")
asyncio.run(main())
```
```python
from deepdiff import DeepDiff, Delta
def calculate_diff(old_data: Dict[str, Any], new_data: Dict[str, Any]) -> Dict[str, Any]:
diff = DeepDiff(old_data, new_data, ignore_order=True).to_dict()
return diff
def apply_diff(data: Dict[str, Any], diff: Dict[str, Any]) -> Dict[str, Any]:
delta = Delta(diff)
delta.apply(data)
return data
```
```python
from odmantic import Model
from pydantic import BaseModel
from typing import Dict, Any
from datetime import datetime
class DocumentBase(BaseModel):
data: Dict[str, Any]
class DocumentCreate(DocumentBase):
pass
class DocumentResponse(DocumentBase):
id: str
latest_version: int
class MyDocument(DocumentBase, Model):
latest_version: int
class DocumentDiff(Model):
document_id: str
version: int
diff: Dict[str, Any]
timestamp: datetime
class Config:
collection = "document_diffs"
```
```python
def recursive_compare(d1, d2, path=""):
diffs = {}
for key in d1.keys():
if key not in d2:
diffs[f"{path}.{key}"] = {"status": "removed", "value": d1[key]}
else:
if isinstance(d1[key], dict) and isinstance(d2[key], dict):
nested_diffs = recursive_compare(d1[key], d2[key], f"{path}.{key}")
diffs.update(nested_diffs)
elif d1[key] != d2[key]:
diffs[f"{path}.{key}"] = {"status": "changed", "old_value": d1[key], "new_value": d2[key]}
for key in d2.keys():
if key not in d1:
diffs[f"{path}.{key}"] = {"status": "added", "value": d2[key]}
return diffs
# Compare the old and new JSON
custom_diff = recursive_compare(old_json, new_json, "root")
print(custom_diff)
```
Example output `recursive_compare`:
```python
d = {"b": {"c": 1, "cc": 2, "ccc": 3}}
d1 = {"W": {"c": 1, "cc": 2, "ccc": 4}}
recursive_compare(d, d1, "root")
{
"root.b": {
"status": "removed",
"value": { "c": 1, "cc": 2, "ccc": 3 }
},
"root.W": {
"status": "added",
"value": { "c": 1, "cc": 2, "ccc": 4}
}
}
```
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment