118 lines
3.1 KiB
Python
118 lines
3.1 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: str
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow, nullable=False, title="Modifié le")
|
|
# updated_by: str
|
|
|
|
@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):
|
|
values = cls.model_validate(create_schema.model_dump()).model_dump(mode="json")
|
|
result = await cls._get_collection(db).insert_one(values)
|
|
|
|
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):
|
|
value = await cls._get_collection(db).find_one({"_id": model_id})
|
|
if not value:
|
|
return None
|
|
|
|
value["id"] = value.pop("_id")
|
|
return cls.model_validate(value)
|
|
|
|
@classmethod
|
|
async def update(cls, db, model, update_schema):
|
|
update_query = {
|
|
"$set": {field: value for field, value in update_schema.model_dump(mode="json").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 = ""
|