|
1 | 1 | from typing import Any, Dict, List, Optional, Type, TypeVar |
2 | 2 |
|
3 | | -import jsonschema |
4 | | -from mat3ra.utils import object as object_utils |
5 | 3 | from pydantic import BaseModel, ConfigDict |
6 | 4 | from pydantic.alias_generators import to_snake |
7 | 5 | from typing_extensions import Self |
8 | 6 |
|
9 | | -from . import BaseUnderscoreJsonPropsHandler |
10 | 7 | from .mixins import DefaultableMixin, HasDescriptionMixin, HasMetadataMixin, NamedMixin |
11 | 8 |
|
12 | 9 | T = TypeVar("T", bound="InMemoryEntityPydantic") |
13 | 10 | B = TypeVar("B", bound="BaseModel") |
14 | 11 |
|
15 | 12 |
|
16 | | -# TODO: remove in the next PR |
17 | | -class ValidationErrorCode: |
18 | | - IN_MEMORY_ENTITY_DATA_INVALID = "IN_MEMORY_ENTITY_DATA_INVALID" |
19 | | - |
20 | | - |
21 | | -# TODO: remove in the next PR |
22 | | -class ErrorDetails: |
23 | | - def __init__(self, error: Optional[Dict[str, Any]], json: Dict[str, Any], schema: Dict): |
24 | | - self.error = error |
25 | | - self.json = json |
26 | | - self.schema = schema |
27 | | - |
28 | | - |
29 | | -# TODO: remove in the next PR |
30 | | -class EntityError(Exception): |
31 | | - def __init__(self, code: ValidationErrorCode, details: Optional[ErrorDetails] = None): |
32 | | - super().__init__(code) |
33 | | - self.code = code |
34 | | - self.details = details |
35 | | - |
36 | | - |
37 | 13 | class InMemoryEntityPydantic(BaseModel): |
38 | 14 | model_config = {"arbitrary_types_allowed": True} |
39 | 15 |
|
@@ -90,82 +66,41 @@ def clone(self: T, extra_context: Optional[Dict[str, Any]] = None, deep=True) -> |
90 | 66 | class InMemoryEntitySnakeCase(InMemoryEntityPydantic): |
91 | 67 | model_config = ConfigDict( |
92 | 68 | arbitrary_types_allowed=True, |
| 69 | + # Generate snake_case aliases for all fields (e.g. myField -> my_field) |
93 | 70 | alias_generator=to_snake, |
| 71 | + # Allow populating fields using either the original name or the snake_case alias |
94 | 72 | populate_by_name=True, |
95 | 73 | ) |
96 | 74 |
|
| 75 | + @staticmethod |
| 76 | + def _create_property_from_camel_case(camel_name: str): |
| 77 | + def getter(self): |
| 78 | + return getattr(self, camel_name) |
97 | 79 |
|
98 | | -# TODO: remove in the next PR |
99 | | -class InMemoryEntity(BaseUnderscoreJsonPropsHandler): |
100 | | - jsonSchema: Optional[Dict] = None |
101 | | - |
102 | | - @classmethod |
103 | | - def get_cls(cls) -> str: |
104 | | - return cls.__name__ |
105 | | - |
106 | | - @property |
107 | | - def cls(self) -> str: |
108 | | - return self.__class__.__name__ |
109 | | - |
110 | | - def get_cls_name(self) -> str: |
111 | | - return self.__class__.__name__ |
112 | | - |
113 | | - @classmethod |
114 | | - def create(cls, config: Dict[str, Any]) -> Any: |
115 | | - return cls(config) |
| 80 | + def setter(self, value: Any): |
| 81 | + setattr(self, camel_name, value) |
116 | 82 |
|
117 | | - def to_json(self, exclude: List[str] = []) -> Dict[str, Any]: |
118 | | - return self.clean(object_utils.clone_deep(object_utils.omit(self._json, exclude))) |
| 83 | + return property(getter, setter) |
119 | 84 |
|
120 | | - def clone(self, extra_context: Dict[str, Any] = {}) -> Any: |
121 | | - config = self.to_json() |
122 | | - config.update(extra_context) |
123 | | - # To avoid: |
124 | | - # Argument 1 to "__init__" of "BaseUnderscoreJsonPropsHandler" has incompatible type "Dict[str, Any]"; |
125 | | - # expected "BaseUnderscoreJsonPropsHandler" |
126 | | - return self.__class__(config) |
| 85 | + def __init_subclass__(cls, **kwargs): |
| 86 | + super().__init_subclass__(**kwargs) |
| 87 | + if not issubclass(cls, BaseModel): |
| 88 | + return |
127 | 89 |
|
128 | | - @staticmethod |
129 | | - def validate_data(data: Dict[str, Any], clean: bool = False): |
130 | | - if clean: |
131 | | - print("Error: clean is not supported for InMemoryEntity.validateData") |
132 | | - if InMemoryEntity.jsonSchema: |
133 | | - jsonschema.validate(data, InMemoryEntity.jsonSchema) |
134 | | - |
135 | | - def validate(self) -> None: |
136 | | - if self._json: |
137 | | - self.__class__.validate_data(self._json) |
138 | | - |
139 | | - def clean(self, config: Dict[str, Any]) -> Dict[str, Any]: |
140 | | - # Not implemented, consider the below for the implementation |
141 | | - # https://stackoverflow.com/questions/44694835/remove-properties-from-json-object-not-present-in-schema |
142 | | - return config |
143 | | - |
144 | | - def is_valid(self) -> bool: |
145 | 90 | try: |
146 | | - self.validate() |
147 | | - return True |
148 | | - except EntityError: |
149 | | - return False |
150 | | - |
151 | | - # Properties |
152 | | - @property |
153 | | - def id(self) -> str: |
154 | | - return self.prop("_id", "") |
| 91 | + model_fields = cls.model_fields |
| 92 | + except Exception: |
| 93 | + return |
155 | 94 |
|
156 | | - @id.setter |
157 | | - def id(self, id: str) -> None: |
158 | | - self.set_prop("_id", id) |
| 95 | + for field_name, field_info in model_fields.items(): |
| 96 | + if field_name == to_snake(field_name): |
| 97 | + continue |
159 | 98 |
|
160 | | - @property |
161 | | - def slug(self) -> str: |
162 | | - return self.prop("slug", "") |
| 99 | + snake_case_name = to_snake(field_name) |
| 100 | + if hasattr(cls, snake_case_name): |
| 101 | + continue |
163 | 102 |
|
164 | | - def get_as_entity_reference(self, by_id_only: bool = False) -> Dict[str, str]: |
165 | | - if by_id_only: |
166 | | - return {"_id": self.id} |
167 | | - else: |
168 | | - return {"_id": self.id, "slug": self.slug, "cls": self.get_cls_name()} |
| 103 | + setattr(cls, snake_case_name, cls._create_property_from_camel_case(field_name)) |
169 | 104 |
|
170 | 105 |
|
171 | 106 | class HasDescriptionHasMetadataNamedDefaultableInMemoryEntityPydantic( |
|
0 commit comments