from datetime import datetime, UTC from typing import Optional from beanie import PydanticObjectId 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): 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 list(cls, db, filters): query = filters.filter(cls._get_collection(db)) query = filters.sort(query) return query @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 = ""