148 lines
4.4 KiB
Python
148 lines
4.4 KiB
Python
from datetime import datetime, UTC
|
|
from typing import Optional, TypedDict, ClassVar
|
|
|
|
from beanie import PydanticObjectId
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
from pydantic import BaseModel, Field, computed_field
|
|
|
|
class CrudDocumentConfig(TypedDict, total=False):
|
|
fulltext_search: list[str]
|
|
indexes: list[str]
|
|
|
|
|
|
class CrudDocument(BaseModel):
|
|
document_config: ClassVar[CrudDocumentConfig] = CrudDocumentConfig()
|
|
|
|
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(title="Label")
|
|
def label(self) -> str:
|
|
return self.compute_label()
|
|
|
|
def compute_label(self) -> str:
|
|
return ""
|
|
|
|
@classmethod
|
|
def _collection_name(cls):
|
|
return cls.__name__
|
|
|
|
@classmethod
|
|
def _get_collection(cls, db) -> AsyncIOMotorCollection:
|
|
return db.get_collection(cls._collection_name())
|
|
|
|
@classmethod
|
|
def create_index(cls, db, index):
|
|
cls._get_collection(db).create_index(index)
|
|
|
|
@classmethod
|
|
async def create(cls, db, create_schema):
|
|
model_dict = create_schema.model_dump() | {"created_by": db.partner.id, "updated_by": db.partner.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
|
|
async def list(cls, db, criteria={}):
|
|
result = []
|
|
for document in await cls._get_collection(db).find(criteria).to_list():
|
|
document["id"] = document.pop("_id")
|
|
result.append(cls.model_validate(document))
|
|
|
|
return result
|
|
|
|
@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.partner.id, "updated_at": datetime.now(UTC)}
|
|
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)
|
|
new_model = await cls.get(db, model.id)
|
|
if new_model.label != model.label:
|
|
await cls._get_collection(db).update_one({"_id": model.id}, {"$set": {"label": new_model.label}})
|
|
return new_model
|
|
|
|
@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)
|
|
|
|
|
|
def ForeignKey(resource, schema, displayed_fields=None, *args, **kwargs):
|
|
kwargs["foreignKey"] = {
|
|
"reference": {
|
|
"resource": resource,
|
|
"schema": schema,
|
|
}
|
|
}
|
|
|
|
if displayed_fields:
|
|
kwargs["foreignKey"]["reference"]["displayedFields"] = displayed_fields
|
|
|
|
return Field(*args, **kwargs)
|
|
|
|
|
|
class DictionaryEntry(BaseModel):
|
|
key: str
|
|
value: str = ""
|