fix when some kind of unions are failed to validate

Author: phenobarbital

datamodel/converters.pyx

@@ -586,6 +586,12 @@ cdef object _parse_typing_type(
     """
     cdef tuple type_args = getattr(T, '__args__', ())
 
+    # print('FIELD > ', field)
+    # print('T > ', T)
+    # print('NAME > ', name)
+    # print('DATA > ', data)
+    # print('TYPE > ', type_args)
+
     if field.origin in {dict, Mapping} or name in {'Dict', 'Mapping'}:
         if isinstance(data, dict):
             if type_args:

datamodel/validation.pyx

@@ -44,13 +44,20 @@ cpdef bool_t is_optional_type(object annotated_type):
     return False
 
 cpdef list _validation(object F, str name, object value, object annotated_type, object val_type, str field_type):
+    cdef bint _valid = False
+
     if not annotated_type:
         annotated_type = F.type
     elif isinstance(annotated_type, Field):
         annotated_type = annotated_type.type
     errors = []
 
     # first: calling (if exists) custom validator:
+    # print('VALIDATION F ', F)
+    # print('VALIDATION NAME ', name)
+    # print('VALIDATION VALUE ', value)
+    # print('VALIDATION ANNOTATED TYPE ', annotated_type)
+
     fn = F.metadata.get('validator', None)
     if fn is not None and callable(fn):
         try:
@@ -153,15 +160,19 @@ cpdef list _validation(object F, str name, object value, object annotated_type,
                 # 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
-                            )
+                    if isinstance(value, base_type):
+                        _valid = True
+                        break
+                if not _valid:
+                    errors.append(
+                        _create_error(
+                            name,
+                            value,
+                            f'Invalid type for {annotated_type}.{name}, expected a type of {inner_types!s}',
+                            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.13'
+__version__ = '0.8.14'
 __copyright__ = 'Copyright (c) 2020-2024 Jesus Lara'
 __author__ = 'Jesus Lara'
 __author_email__ = '[email protected]'

examples/test_driver.py

@@ -1,7 +1,125 @@
 from typing import Union, Optional
-from dataclasses import InitVar
+from dataclasses import asdict, InitVar
 from pathlib import Path
 from datamodel import BaseModel, Field
+from datamodel.exceptions import ValidationError
+
+
+def default_properties() -> tuple:
+    return ('host', 'port', 'user', 'username', 'password')
+
+class BaseDriver(BaseModel):
+    """BaseDriver.
+
+    Description: Base class for all required datasources.
+    """
+    driver: str = Field(required=True, primary_key=True)
+    driver_type: str = Field(
+        required=True,
+        default='asyncdb',
+        comment="type of driver, can be asyncdb, qs or REST"
+    )
+    name: str = Field(required=False, comment='Datasource name, default to driver.')
+    description: str = Field(comment='Datasource Description', repr=False)
+    icon: str = Field(required=False, comment='Icon Path for Datasource.', repr=False)
+    dsn: str = Field(default=None)
+    dsn_format: str = Field(required=False, default=None, repr=False)
+    user: InitVar = Field(default='')
+    username: str = Field(default='')
+    password: str = Field(required=False, default=None, repr=False, is_secret=True)
+    auth: dict = Field(required=False, default_factory=dict)
+    required_properties: Optional[tuple] = Field(
+        repr=False,
+        default=default_properties(),
+        default_factory=tuple
+    )
+
+    def __post_init__(self, user, **kwargs) -> None:  # pylint: disable=W0613,W0221
+        if not self.name:
+            self.name = self.driver
+        if user:
+            self.username = user
+        self.auth = {
+            "username": self.username,
+            "password": self.password
+        }
+        # set DSN (if needed)
+        if self.dsn_format is not None and self.dsn is None:
+            self.create_dsn()
+        super(BaseDriver, self).__post_init__()
+
+    def create_dsn(self) -> str:
+        """create_dsn.
+
+        Description: creates DSN from DSN Format.
+        Returns:
+            str: DSN.
+        """
+        params = asdict(self)
+        try:
+            self.dsn = self.dsn_format.format(**params)
+            return self.dsn
+        except (AttributeError, ValueError):
+            return None
+
+    def get_credentials(self) -> dict:
+        """get_credentials.
+
+        Description: Returns credentials for Datasource.
+        Returns:
+            dict: credentials.
+        """
+        return self.params()
+
+    def get_parameters(self) -> dict:
+        return {}
+
+    @classmethod
+    def properties(cls) -> dict:
+        """properties.
+
+        Description: Returns fields related to Drivers Supported.
+        Returns:
+            dict: all required fields for Supported Drivers.
+        """
+
+        fields = {}
+        for field in cls.required_properties:
+            # because tuple is ordered:
+            try:
+                f = cls.column(cls, field)
+            except KeyError:
+                continue  # Field Missing on Driver:
+            secret = False
+            if 'is_secret' in f.metadata:
+                secret = f.metadata["is_secret"]
+            title = field
+            if 'title' in f.metadata:
+                title = f.metadata['title']
+            required = False
+            if 'required' in f.metadata:
+                required = f.metadata['required']
+            f = {
+                "name": field,
+                "title": title,
+                "required": required,
+                "is_secret": secret
+            }
+            value = getattr(cls, field)
+            default = hasattr(f, 'default')
+            if not value and default:
+                value = f.default
+            if value:
+                f["value"] = value
+            fields[field] = f
+        return {
+            "driver": cls.driver,
+            "name": cls.name,
+            "icon": cls.icon,
+            "dsn_format": cls.dsn_format,
+            "fields": fields
+        }
+
 
 def jdbc_properties() -> tuple:
     return ('host', 'port', 'user', 'password', 'database', 'dsn', 'jar', 'classpath')
@@ -20,7 +138,7 @@ class jdbcDriver(BaseModel):
     jar: Union[list, str] = Field(Required=True)
     classpath: Path = Field(Required=False)
     required_properties: Optional[Union[list, tuple]] = Field(
-        repr=False, default=jdbc_properties()
+        repr=False, default=jdbc_properties(), default_factory=tuple
     )
 
     def __post_init__(self, username, *args, **kwargs):
@@ -53,9 +171,21 @@ def params(self) -> dict:
         jar='/Users/jlara/.m2/repository/com/oracle/ojdbc/ojdbc8/',
         classpath='/Users/jlara/.m2/repository/com/oracle/ojdbc/ojdbc8/ojdbc8-'
     )
+    print('JDBC > ', jdbc_default)
 except ValueError:
     jdbc_default = None
 except Exception as e:
     print('ERROR > ', e, type(e))
     print('PAYLOAD > ', e.payload)
     jdbc_default = None
+
+try:
+    base_driver = BaseDriver(
+        driver='asyncdb',
+        user='admin',
+        password='admin',
+        name='asyncdb_test'
+    )
+    print('BASE DRIVER > ', base_driver)
+except ValidationError as e:
+    print(e.payload)