Merge pull request #229 from phenobarbital/bugfix-optional-list-str

supporting new converters and validations

Author: phenobarbital

datamodel/abstract.py

@@ -1,5 +1,6 @@
 import logging
 from typing import Optional, Any, List, Dict, get_args, get_origin
+from types import GenericAlias
 from collections import OrderedDict
 from collections.abc import Callable
 import types
@@ -190,7 +191,8 @@ def _initialize_fields(attrs, annotations, strict):
                     _default = df.default
                     _is_dc = is_dataclass(_type)
                     _is_prim = is_primitive(_type)
-                    _is_typing = hasattr(_type, '__module__') and _type.__module__ == 'typing'
+                    _is_alias = isinstance(_type, GenericAlias)
+                    _is_typing = hasattr(_type, '__module__') and _type.__module__ == 'typing'  # noqa
 
                     # Store the type info in the field object:
                     df.is_dc = _is_dc
@@ -207,12 +209,16 @@ def _initialize_fields(attrs, annotations, strict):
                     # check type of field:
                     if _is_prim:
                         _type_category = 'primitive'
+                    elif origin == type:
+                        _type_category = 'typing'
                     elif _is_dc:
                         _type_category = 'dataclass'
                     elif _is_typing:  # noqa
                         _type_category = 'typing'
                     elif isclass(_type):
                         _type_category = 'class'
+                    elif _is_alias:
+                        _type_category = 'typing'
                     else:
                         _type_category = 'complex'
                     _types_local[field] = _type_category

datamodel/converters.pyx

@@ -3,6 +3,7 @@
 #
 import re
 from typing import get_args, get_origin, Union, Optional, List
+from collections.abc import Sequence, Mapping, Callable, Awaitable
 from dataclasses import _MISSING_TYPE, _FIELDS, fields
 import ciso8601
 import orjson
@@ -446,22 +447,6 @@ cdef object _parse_dict_type(
         new_dict[k] = parse_typing(field, val_type, v, encoder, False)
     return new_dict
 
-cdef object _unwrap_optional(object T):
-    """
-    If T is a Union that includes NoneType (i.e. Optional[T]),
-    return the non-None type, else T unchanged.
-    """
-    cdef object orig = get_origin(T)
-    cdef tuple args = None
-    cdef list non_none = []
-    if orig is Union:
-        args = get_args(T)
-        # If exactly one type is not NoneType, return it.
-        non_none = [a for a in args if a is not type(None)]
-        if len(non_none) == 1:
-            return non_none[0]
-    return T, args
-
 cdef object _parse_list_type(
     object field,
     object T,
@@ -601,13 +586,30 @@ cdef object _parse_typing_type(
     """
     cdef tuple type_args = getattr(T, '__args__', ())
 
-    if name == 'Dict' and isinstance(data, dict):
-        if type_args:
-            # e.g. Dict[K, V]
-            return {k: _parse_type(field, type_args[1], v, None, False) for k, v in data.items()}
-        return data
+    if field.origin in {dict, Mapping} or name in {'Dict', 'Mapping'}:
+        if isinstance(data, dict):
+            if type_args:
+                # e.g. Dict[K, V]
+                return {k: _parse_type(field, type_args[1], v, None, False) for k, v in data.items()}
+            return data
+
+    if name == 'Tuple' or field.origin == tuple:
+        if isinstance(data, (list, tuple)):
+            if len(data) == len(type_args):
+                return tuple(
+                    _parse_type(field, typ, datum, encoder, False)
+                    for typ, datum in zip(type_args, data)
+                )
+            else:
+                if len(type_args) == 2 and type_args[1] is Ellipsis:
+                    # e.g. Tuple[str, ...]
+                    return tuple(
+                        _parse_type(field, type_args[0], datum, None, False)
+                        for datum in data
+                    )
+        return tuple(data)
 
-    if name == 'List':
+    if name in {'List', 'Sequence'} or field.origin in {list, Sequence}:
         if not isinstance(data, (list, tuple)):
             data = [data]
         return _parse_list_typing(

datamodel/functions.pyx

@@ -68,6 +68,10 @@ cpdef bool_t is_empty(object value):
         result = True
     elif isinstance(value, (int, float)) and value == 0:
         result = False
+    elif isinstance(value, dict) and value == {}:
+        result = False
+    elif isinstance(value, (list, tuple, set)) and value == []:
+        result = False
     elif not value:
         result = True
     return result

datamodel/validation.pyx

@@ -2,7 +2,9 @@
 # Copyright (C) 2018-present Jesus Lara
 #
 from typing import get_args, get_origin, Union, Optional
+from collections.abc import Callable, Awaitable
 import typing
+import asyncio
 import inspect
 from libcpp cimport bool as bool_t
 from enum import Enum
@@ -50,20 +52,19 @@ cpdef list _validation(object F, str name, object value, object annotated_type,
 
     # first: calling (if exists) custom validator:
     fn = F.metadata.get('validator', None)
-    if fn is not None:
-        if is_callable(fn):
-            try:
-                result = fn(F, value)
-                if not result:
-                    error_msg = f"Validator {fn!r} Failed: {result}"
-                    errors.append(
-                        _create_error(name, value, error_msg, val_type, annotated_type)
-                    )
-            except (ValueError, AttributeError, TypeError) as e:
+    if fn is not None and callable(fn):
+        try:
+            result = fn(F, value)
+            if not result:
                 error_msg = f"Validator {fn!r} Failed: {result}"
                 errors.append(
-                    _create_error(name, value, error_msg, val_type, annotated_type, e)
+                    _create_error(name, value, error_msg, val_type, annotated_type)
                 )
+        except (ValueError, AttributeError, TypeError) as e:
+            error_msg = f"Validator {fn!r} Failed: {result}"
+            errors.append(
+                _create_error(name, value, error_msg, val_type, annotated_type, e)
+            )
     # check: data type hint
     try:
         # If field_type is known, short-circuit certain checks
@@ -88,8 +89,74 @@ cpdef list _validation(object F, str name, object value, object annotated_type,
                 errors.append(
                     _create_error(name, value, f'invalid type for {annotated_type}.{name}, expected {annotated_type}', val_type, annotated_type)
                 )
-        elif hasattr(annotated_type, '__module__') and annotated_type.__module__ == 'typing':
-            # TODO: validation of annotated types
+        elif F.origin is Callable:
+            if not is_callable(value):
+                errors.append(
+                    _create_error(name, value, f'Invalid function type, expected {annotated_type}', val_type, annotated_type)
+                )
+        elif F.origin is Awaitable:
+            if asyncio.iscoroutinefunction(value):
+                errors.append(
+                    f"Field '{name}': provided coroutine function is not awaitable; call it to obtain a coroutine object."
+                )
+            # Otherwise, check if it is awaitable
+            elif not hasattr(value, '__await__'):
+                errors.append(
+                    f"Field '{name}': expected an awaitable, but got {type(value)}."
+                )
+        elif field_type == 'typing' or hasattr(annotated_type, '__module__') and annotated_type.__module__ == 'typing':
+            if F.origin is type:
+                for allowed in F.args:
+                    if isinstance(value, allowed):
+                        break
+                else:
+                    expected = ', '.join([str(t) for t in F.args])
+                    errors.append(
+                        _create_error(name, value, f'Invalid type for {annotated_type}.{name}, expected a subclass of {expected}', val_type, annotated_type)
+                    )
+            if F.origin is tuple:
+                # Check if we are in the homogeneous case: Tuple[T, ...]
+                if len(F.args) == 2 and F.args[1] is Ellipsis:
+                    for i, elem in enumerate(value):
+                        if not isinstance(elem, F.args[0]):
+                            errors.append(
+                                _create_error(f"{name}[{i}]", elem,
+                                    f"Invalid type at index {i}: expected {F.args[0]}",
+                                    type(elem), F.args[0])
+                            )
+                else:
+                    if len(value) != len(F.args):
+                        errors.append(
+                            _create_error(name, value,
+                                f"Invalid number of elements: expected {len(F.args)}, got {len(value)}",
+                                len(value), len(F.args))
+                        )
+                    else:
+                        for i, elem in enumerate(value):
+                            if not isinstance(elem, F.args[i]):
+                                errors.append(
+                                    _create_error(f"{name}[{i}]", elem,
+                                        f"Invalid type at index {i}: expected {F.args[i]}",
+                                        type(elem), F.args[i])
+                                )
+            # Handle Optional Types:
+            elif F.origin is Union and type(None) in F.args:
+                inner_types = [t for t in F.args if t is not type(None)]
+                # If value is None then that is valid:
+                if value is None:
+                    return errors
+                # Otherwise check that value is an instance of at least one inner type:
+                for t in inner_types:
+                    base_type = get_origin(t) or t
+                    if not isinstance(value, base_type):
+                        errors.append(
+                            _create_error(
+                                name,
+                                value,
+                                f"Invalid type for Optional field; expected one of {inner_types}",
+                                val_type, annotated_type
+                            )
+                        )
             return errors
         # elif type(annotated_type) is ModelMeta:
         elif type(annotated_type).__name__ == "ModelMeta":

datamodel/version.py

@@ -6,7 +6,7 @@
     'simple library based on python +3.8 to use Dataclass-syntax'
     'for interacting with Data'
 )
-__version__ = '0.8.12'
+__version__ = '0.8.13'
 __copyright__ = 'Copyright (c) 2020-2024 Jesus Lara'
 __author__ = 'Jesus Lara'
 __author_email__ = '[email protected]'

examples/test_callables.py

@@ -0,0 +1,30 @@
+# test_example_model.py
+from typing import Callable, Awaitable
+import asyncio
+import pytest
+from datamodel import BaseModel, Field
+from datamodel.exceptions import ValidationError
+
+# --- Valid Cases ---
+
+# Define a simple synchronous callback function.
+def my_callback(x: int) -> str:
+    return f"Value: {x}"
+
+# Define an asynchronous function.
+async def my_async_func() -> int:
+    await asyncio.sleep(0.01)
+    return 42
+
+# Define the ExampleModel using your DataModel framework:
+class ExampleModel(BaseModel):
+    callback: Callable[[int], str] = Field(required=True)
+    async_result: Awaitable[int] = Field(required=True)
+
+try:
+    instance = ExampleModel(
+        callback=my_callback,
+        async_result=my_async_func()  # note: a coroutine object
+    )
+except ValidationError as e:
+    print(e.payload)

examples/test_qsmodel.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List, Optional, Mapping, Sequence, Callable, Tuple, Union
 from datetime import datetime
 from datamodel import BaseModel, Field
 from datamodel.exceptions import ValidationError
@@ -9,8 +9,11 @@
     'description': 'walmart_mtd_postpaid_to_goal',
     'conditions': {'filterdate': 'POSTPAID_DATE', 'store_tier': 'null', 'launch_group': 'null'},
     'cond_definition': {'filterdate': 'date', 'store_tier': 'string', 'launch_group': 'string'},
+    'attributes': {"example": "value"},
     'fields': ["client_id", "client_name"], 'ordering': [],
     'h_filtering': False,
+    'directives': ('23.1', 12.8),
+    'supported': ('1.0', 2.0, '3.0'),
     'query_raw': 'SELECT {fields}\nFROM walmart.postpaid_metrics({filterdate}, {launch_group}, {store_tier})\n{where_cond}',
     'is_raw': False,
     'is_cached': True,
@@ -22,7 +25,11 @@
     'program_id': 3,
     'program_slug': 'walmart', 'dwh': False, 'dwh_scheduler': {},
     'created_at': datetime(2022, 11, 18, 1, 10, 8, 872163),
-    'updated_at': datetime(2023, 4, 18, 1, 38, 44, 466221)
+    'updated_at': datetime(2023, 4, 18, 1, 38, 44, 466221),
+    'host_info': ('host', 4568.1),
+    'created_by': (1, 33),
+    "example": {"a": "hello", "b": 123},
+    "more_nested": {"a": "hello", "b": 123}
 }
 
 def rigth_now(obj) -> datetime:
@@ -57,24 +64,29 @@ class QueryModel(BaseModel):
     ## Program Information:
     program_id: int = Field(required=True, default=1)
     program_slug: str = Field(required=True, default='default')
+    directives: Tuple[float, float] = Field(required=False)
+    supported: Tuple[float, ...] = Field(required=False)
     # DWH information
     dwh: bool = Field(required=True, default=False)
     dwh_driver: str = Field(required=False, default=None)
     dwh_info: Optional[dict] = Field(required=False, db_type='jsonb')
     dwh_scheduler: Optional[dict] = Field(required=False, db_type='jsonb')
+    host_info: Tuple[str, int] = Field(required=True)
     # Creation Information:
     created_at: datetime = Field(
         required=False,
         default=datetime.now,
         db_default='now()'
     )
-    created_by: int = Field(required=False)  # TODO: validation for valid user
+    created_by: tuple[int, str] = Field(required=False)  # TODO: validation for valid user
     updated_at: datetime = Field(
         required=False,
         default=datetime.now,
         encoder=rigth_now
     )
     updated_by: int = Field(required=False)  # TODO: validation for valid user
+    example: Mapping[str, str | int] = Field(required=False)
+    more_nested: Mapping[str, Union[str, str]] = Field(required=False)
 
     class Meta:
         driver = 'pg'

examples/test_type_user.py

@@ -0,0 +1,41 @@
+from typing import Type, Union
+from datamodel import BaseModel, Field
+from datamodel.exceptions import ValidationError
+
+# Define a basic user hierarchy.
+class User(BaseModel):
+    username: str = Field(required=True)
+    email: str = Field(required=True)
+
+class BasicUser(User):
+    level: str = Field(default="basic")
+
+class ProUser(User):
+    level: str = Field(default="pro")
+    perks: list = Field(default_factory=list)
+
+# Employee model whose user_class field must be a type (class)
+# that is either BasicUser or ProUser.
+class Employee(BaseModel):
+    # The type hint below means: user_class must be a type (class)
+    # that is a subclass of either BasicUser or ProUser.
+    user_class: type[BasicUser | ProUser] = Field(required=True)
+
+
+try:
+    # This should raise a ValidationError because the type hint
+    # specifies that user_class must be a type (class), not an instance.
+    user = BasicUser(username="user", email="email")
+    employee = Employee(user_class=user)
+    print(employee)
+except ValidationError as e:
+    print(e.payload)
+
+try:
+    # This should raise a ValidationError because the type hint
+    # specifies that user_class must be a type (class), not an instance.
+    user = User(username="user", email="email")
+    employee = Employee(user_class=user)
+    print(employee)
+except ValidationError as e:
+    print(e.payload)