fix bug when a user is passed as object to a datamodel

Author: phenobarbital

datamodel/base.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from typing import Any, Optional
 import inspect
 import logging
@@ -22,6 +23,9 @@
 from .models import ModelMixin
 
 
+TYPE_CONVERTERS = {}  # Maps a type to a conversion callable
+
+
 class BaseModel(ModelMixin, metaclass=ModelMeta):
     """
     BaseModel.
@@ -106,7 +110,12 @@ def _handle_default_value(self, value, f, name) -> Any:
         # Otherwise, return value as-is
         return value
 
-    def _handle_dataclass_type(self, value, _type):
+    @classmethod
+    def register_converter(cls, target_type: Any, func: Callable, field_name: str = None):
+        key = (target_type, field_name) if field_name else target_type
+        TYPE_CONVERTERS[key] = func
+
+    def _handle_dataclass_type(self, name: str, value: Any, _type: Any):
         try:
             if hasattr(self.Meta, 'no_nesting'):
                 return value
@@ -116,21 +125,47 @@ def _handle_dataclass_type(self, value, _type):
                 return _type(**value)
             if isinstance(value, list):
                 return _type(*value)
-            return value if isinstance(value, (int, str, UUID)) else _type(value)
+            else:
+                # If a converter exists for this type, use it:
+                key = (_type, name)
+                converter = TYPE_CONVERTERS.get(key) or TYPE_CONVERTERS.get(_type)
+                if converter:
+                    return converter(name, value, _type)
+                if getattr(self.Meta, 'as_objects', False) is True:
+                    return _type(**{name: value})
+                if isinstance(value, (int, str, UUID)):
+                    return value
+                if inspect.isclass(_type) and hasattr(_type, '__dataclass_fields__'):
+                    return _type(**{name: value})
+                else:
+                    return _type(value)
         except Exception as exc:
             raise ValueError(
                 f"Invalid value for {_type}: {value}, error: {exc}"
             )
 
-    def _handle_list_of_dataclasses(self, value, _type):
+    def _handle_list_of_dataclasses(self, name: str, value: Any, _type: Any):
+        """
+        _handle_list_of_dataclasses.
+
+        Process a list field that is annotated as List[SomeDataclass].
+        If there's a registered converter for the sub-dataclass, call it;
+        otherwise, build the sub-dataclass using default logic.
+        """
         try:
             sub_type = _type.__args__[0]
             if is_dataclass(sub_type):
-                return [
-                    sub_type(
-                        **item
-                    ) if isinstance(item, dict) else item for item in value
-                ]
+                key = (sub_type, name)
+                converter = TYPE_CONVERTERS.get(key) or TYPE_CONVERTERS.get(_type)
+                new_list = []
+                for item in value:
+                    if converter:
+                        new_list.append(converter(name, item, sub_type))
+                    elif isinstance(item, dict):
+                        new_list.append(sub_type(**item))
+                    else:
+                        new_list.append(item)
+                return new_list
         except AttributeError:
             pass
         return value
@@ -160,13 +195,13 @@ def _process_field_(
                 new_val = parse_basic(_type, value, _encoder)
                 return self._validation_(name, new_val, f, _type)
             elif field_category == 'dataclass':
-                new_val = self._handle_dataclass_type(value, _type)
+                new_val = self._handle_dataclass_type(name, value, _type)
                 return self._validation_(name, new_val, f, _type)
             elif field_category == 'typing':
                 new_val = parse_type(_type, value, _encoder, field_category)
                 return self._validation_(name, new_val, f, _type)
             elif isinstance(value, list) and hasattr(_type, '__args__'):
-                new_val = self._handle_list_of_dataclasses(value, _type)
+                new_val = self._handle_list_of_dataclasses(name, value, _type)
                 return self._validation_(name, new_val, f, _type)
             else:
                 new_val = parse_type(f.type, value, _encoder, field_category)

datamodel/validation.pyx

@@ -79,20 +79,27 @@ cpdef list _validation(object F, str name, object value, object annotated_type,
                 )
         elif hasattr(annotated_type, '__module__') and annotated_type.__module__ == 'typing':
             # TODO: validation of annotated types
-            pass
+            return errors
         elif type(annotated_type) is ModelMeta:
             # Check if there's a field in the annotated type that matches the name and type
+            if isinstance(value, annotated_type):
+                # if value is already a User, no further check needed for columns
+                return errors
             try:
                 field = annotated_type.get_column(name)
-                field_type = field.type
-                if field_type <> val_type:
+                ftype = field.type
+                if ftype <> val_type:
                     errors.append(
-                        _create_error(name, value, f'invalid type for {annotated_type}.{name}, expected {field_type}', val_type, annotated_type)
+                        _create_error(name, value, f'invalid type for {annotated_type}.{name}, expected {ftype}', val_type, annotated_type)
                     )
             except AttributeError as e:
                 errors.append(
                     _create_error(name, value, f'{annotated_type} has no column {name}', val_type, annotated_type, e)
                 )
+            except Exception as e:
+                errors.append(
+                    _create_error(name, value, f'Error validating {annotated_type}.{name}', val_type, annotated_type, e)
+                )
         elif is_optional_type(annotated_type):
             inner_types = get_args(annotated_type)
             for t in inner_types:

datamodel/version.py

@@ -4,7 +4,7 @@
 __title__ = 'python-datamodel'
 __description__ = ('simple library based on python +3.8 to use Dataclass-syntax'
                    'for interacting with Data')
-__version__ = '0.8.0'
+__version__ = '0.8.1'
 __author__ = 'Jesus Lara'
 __author_email__ = '[email protected]'
 __license__ = 'BSD'

examples/validate_user.py

@@ -0,0 +1,91 @@
+from datetime import datetime
+from typing import Any
+from datamodel import BaseModel, Column
+from datamodel.exceptions import ValidationError
+
+
+class User(BaseModel):
+    user_id: int = Column(required=True, primary_key=True)
+    name: str = Column(required=False)
+    email: str = Column(required=False)
+    created_at: datetime = Column(required=False, default=datetime.now())
+    created_by: str = Column(required=False)
+
+    class Meta:
+        name = "users"
+        schema = "public"
+        strict = True
+        connection = None
+        frozen = False
+
+class Group(BaseModel):
+    group_id: int = Column(required=True, primary_key=True)
+    name: str = Column(required=True)
+    created_at: datetime = Column(required=False, default=datetime.now())
+    created_by: str = Column(required=False)
+
+    class Meta:
+        name = "groups"
+        schema = "public"
+        strict = True
+        connection = None
+        frozen = False
+
+class UserGroup(BaseModel):
+    user_id: User = Column(required=True, primary_key=True)
+    group_id: Group = Column(required=True, primary_key=True)
+    created_at: datetime = Column(required=False, default=datetime.now())
+    created_by: str = Column(required=False)
+
+    class Meta:
+        name = "user_groups"
+        schema = "public"
+        strict = True
+        connection = None
+        frozen = False
+
+
+class UserAttributes(BaseModel):
+    user_id: User = Column(required=True, primary_key=True)
+    attributes: dict = Column(required=False, default_factory=dict)
+
+    class Meta:
+        name = "users_attributes"
+        schema = "public"
+        strict = True
+        connection = None
+        frozen = False
+
+
+# Path: examples/validate_user.py
+try:
+    user = User(user_id=1, name="user1", email="email1")
+    print(user)
+except ValidationError as e:
+    print(e.payload)
+
+def create_user(name: str, value: Any, target_type: Any):
+    print('Target: ', target_type, value, name)
+    args = {
+        name: value,
+        "name": "John Doe",
+        "created_by": "John Doe",
+    }
+    return target_type(**args)
+
+BaseModel.register_converter(User, create_user, 'user_id')
+
+try:
+    user_attributes = UserAttributes(user_id=1, attributes={"key": "value"})
+    print(user_attributes)
+    print('export user attributes')
+    print(user_attributes.to_json())
+except ValidationError as e:
+    print(e.payload)
+
+# UserAttributes.Meta.as_objects = True
+# try:
+#     user_attributes = UserAttributes(user_id=1, attributes={"key": "value"})
+#     print(user_attributes)
+# except ValidationError as e:
+#     print(e.payload)

tests/test_qsobject.py

@@ -45,7 +45,6 @@ class QueryObject(ClassDict):
 
 def test_queryobject():
     qry = QueryObject(**data)
-    print('AQUI >> ', qry.fields)
     assert qry.fields == ["store_id", "postpaid_sales", "postpaid_trended", "apd", "postpaid_to_goal", "hours_worked", "hps", "hps_to_goal"]
     fields = qry.pop('fields')
     assert fields == ["store_id", "postpaid_sales", "postpaid_trended", "apd", "postpaid_to_goal", "hours_worked", "hps", "hps_to_goal"]