120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
from datetime import datetime, UTC
|
|
from typing import Optional
|
|
|
|
from beanie import PydanticObjectId
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
from pydantic import BaseModel, Field, computed_field
|
|
|
|
|
|
class CrudDocument(BaseModel):
|
|
id: Optional[PydanticObjectId] = Field(default=None)
|
|
created_at: datetime = Field(default=datetime.now(UTC), nullable=False, title="Créé le")
|
|
created_by: Optional[PydanticObjectId] = Field(default=None, title="Créé par")
|
|
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC), nullable=False, title="Modifié le")
|
|
updated_by: Optional[PydanticObjectId] = Field(default=None, title="Modifié par")
|
|
|
|
@property
|
|
def _id(self):
|
|
return self.id
|
|
|
|
@computed_field
|
|
def label(self) -> str:
|
|
return self.compute_label()
|
|
|
|
def compute_label(self) -> str:
|
|
return ""
|
|
|
|
class Settings:
|
|
fulltext_search = []
|
|
|
|
@classmethod
|
|
def _collection_name(cls):
|
|
return cls.__name__
|
|
|
|
@classmethod
|
|
def _get_collection(cls, db) -> AsyncIOMotorCollection:
|
|
return db.get_collection(cls._collection_name())
|
|
|
|
@classmethod
|
|
async def create(cls, db, create_schema):
|
|
model_dict = create_schema.model_dump() | {"created_by": db.user.id, "updated_by": db.user.id}
|
|
document = cls.model_validate(model_dict).model_dump(mode="json")
|
|
result = await cls._get_collection(db).insert_one(document)
|
|
|
|
return await cls.get(db, result.inserted_id)
|
|
|
|
@classmethod
|
|
def find(cls, db, filters):
|
|
return {
|
|
"collection": cls._get_collection(db),
|
|
"query_filter": filters.filter({}),
|
|
"sort": filters.sort(),
|
|
}
|
|
|
|
@classmethod
|
|
def list(cls, db):
|
|
return cls._get_collection(db).find({})
|
|
|
|
@classmethod
|
|
async def get(cls, db, model_id):
|
|
document = await cls._get_collection(db).find_one({"_id": model_id})
|
|
if not document:
|
|
return None
|
|
|
|
document["id"] = document.pop("_id")
|
|
return cls.model_validate(document)
|
|
|
|
@classmethod
|
|
async def update(cls, db, model, update_schema):
|
|
model_dict = update_schema.model_dump(mode="json") | {"updated_by": db.user.id}
|
|
update_query = {
|
|
"$set": {field: value for field, value in model_dict.items() if field!= "id" }
|
|
}
|
|
|
|
await cls._get_collection(db).update_one({"_id": model.id}, update_query)
|
|
return await cls.get(db, model.id)
|
|
|
|
@classmethod
|
|
async def delete(cls, db, model):
|
|
await cls._get_collection(db).delete_one({"_id": model.id})
|
|
|
|
|
|
|
|
def text_area(*args, **kwargs):
|
|
kwargs['widget'] = {
|
|
"formlyConfig": {
|
|
"type": "textarea",
|
|
"props": {
|
|
"placeholder": "Leaving this field empty will cause formData property to be `null`",
|
|
"rows": kwargs['size'] if 'size' in kwargs else 10
|
|
}
|
|
}
|
|
}
|
|
|
|
return Field(*args, **kwargs)
|
|
|
|
|
|
def RichtextMultiline(*args, **kwargs):
|
|
if 'props' not in kwargs:
|
|
kwargs['props'] = {}
|
|
|
|
kwargs['props']['richtext'] = True
|
|
kwargs['props']['multiline'] = True
|
|
|
|
return Field(*args, **kwargs)
|
|
|
|
|
|
def RichtextSingleline(*args, **kwargs):
|
|
if 'props' not in kwargs:
|
|
kwargs['props'] = {}
|
|
|
|
kwargs['props']['richtext'] = True
|
|
kwargs['props']['multiline'] = False
|
|
|
|
return Field(*args, **kwargs)
|
|
|
|
|
|
class DictionaryEntry(BaseModel):
|
|
key: str
|
|
value: str = ""
|