Merge pull request #249 from phenobarbital/new-exports

added validation for typing.Literal

Author: phenobarbital

datamodel/converters.pyx

@@ -2,7 +2,7 @@
 # Copyright (C) 2018-present Jesus Lara
 #
 import re
-from typing import get_args, get_origin, Union, Optional, List, NewType
+from typing import get_args, get_origin, Union, Optional, List, NewType, Literal
 from collections.abc import Sequence, Mapping, Callable, Awaitable
 import types
 from dataclasses import _MISSING_TYPE, _FIELDS, fields
@@ -80,13 +80,21 @@ cpdef str to_string(object obj):
             return obj.decode()
         except UnicodeDecodeError as e:
             raise ValueError(f"Cannot decode bytes: {e}") from e
+    if isinstance(obj, (int, float, Decimal)):
+        # its a number
+        return str(obj)
     if callable(obj):
         # its a function callable returning a value
         try:
-            return str(obj())
+            val = obj()
+            # Recursively call to_string on that result:
+            return to_string(val)
         except Exception:
             pass
-    return str(obj)
+    # For any other arbitrary type, explicitly fail:
+    raise ValueError(
+        f"Cannot convert object of type {type(obj).__name__} to string."
+    )
 
 cpdef object to_uuid(object obj):
     """Returns a UUID version of a str column.
@@ -1123,6 +1131,36 @@ cdef object _parse_typing(
         # fallback to builtin parse
         return _parse_builtin_type(field, T, data, encoder)
 
+cdef object _parse_literal_type(
+    object field,
+    object T,
+    object data,
+    object encoder
+):
+    """
+    _parse_literal_type parses a typing.Literal[...] annotation.
+
+    :param field: A Field object (or similar) containing metadata
+    :param T: The full annotated type (e.g. typing.Literal['text/plain', 'text/html']).
+    :param data: The input value to check.
+    :param encoder: Optional encoder (not usually used for literal).
+    :return: Returns 'data' if it matches one of the literal choices, otherwise raises ValueError.
+    """
+
+    # Each element in `targs` is a valid literal value, e.g. a string, int, etc.
+    # If data is exactly in that set, it's valid.
+    cdef tuple targs = field.args
+    cdef tuple i
+    for arg in targs:
+        if data == arg:
+            return data
+
+    # If we get here, data didn't match any literal value
+    raise ValueError(
+        f"Literal parse error for field '{field.name}': "
+        f"value={data!r} is not one of {targs}"
+    )
+
 cdef object _handle_dataclass_type(
     object field,
     str name,
@@ -1386,6 +1424,10 @@ cpdef dict processing_fields(object obj, list columns):
                     # means that is_dataclass(T)
                     newval = _handle_dataclass_type(None, name, value, _type, as_objects, None)
                     obj.__dict__[name] = newval
+                elif f.origin is Literal:
+                    # e.g. Literal[...]
+                    newval = _parse_literal_type(f, _type, value, _encoder)
+                    obj.__dict__[name] = newval
                 elif f.origin is list:
                     # Other typical case is when is a List of primitives.
                     if f._inner_priv:

datamodel/validation.pyx

@@ -1,7 +1,7 @@
 # cython: language_level=3, embedsignature=True, initializedcheck=False
 # Copyright (C) 2018-present Jesus Lara
 #
-from typing import get_args, get_origin, Union, Optional
+from typing import get_args, get_origin, Union, Optional, Literal
 from collections.abc import Callable, Awaitable
 import typing
 import asyncio
@@ -263,6 +263,7 @@ cpdef dict _validation(
     cdef bint _valid = False
     cdef object field_meta = F.metadata
     cdef dict error = {}
+    cdef list allowed_values = []
 
     if not annotated_type:
         annotated_type = F.type
@@ -307,6 +308,16 @@ cpdef dict _validation(
         return {}
     elif field_type == 'type':
         return validate_type(F, name, value, annotated_type, val_type)
+    elif F.origin is Literal:
+        allowed_values = list(F.args)
+        if value not in allowed_values:
+            return _create_error(
+                name,
+                value,
+                f"Invalid value for {annotated_type}.{name}, expected one of {allowed_values}",
+                val_type,
+                annotated_type
+            )
     elif field_type == 'typing' or hasattr(annotated_type, '__module__') and annotated_type.__module__ == 'typing':
         if F.origin is tuple:
             # Check if we are in the homogeneous case: Tuple[T, ...]

datamodel/version.py

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

examples/test_notify.py

@@ -0,0 +1,118 @@
+from typing import Union, Optional, Any, Literal
+import uuid
+from datetime import datetime
+from pathlib import Path
+from datamodel import BaseModel, Field
+
+
+CONTENT_TYPES = [
+    "text/plain",
+    "text/html",
+    "multipart/alternative",
+    "application/json"
+]
+
+def auto_uuid(*args, **kwargs):  # pylint: disable=W0613
+    return uuid.uuid4()
+
+
+def now():
+    return datetime.now()
+
+class Account(BaseModel):
+    """
+    Attributes for using a Provider by an User (Actor)
+    """
+
+    provider: str = Field(required=True, default="dummy")
+    enabled: bool = Field(required=True, default=True)
+    address: Union[str, list[str]] = Field(required=False, default_factory=list)
+    number: Union[str, list[str]] = Field(required=False, default_factory=list)
+    userid: str = Field(required=False, default="")
+    attributes: dict = Field(required=False, default_factory=dict)
+
+    def set_address(self, address: Union[str, list[str]]):
+        self.address = [address] if isinstance(address, str) else address
+
+
+class Actor(BaseModel):
+    """
+    Basic Actor (meta-definition), can be an Sender or a Recipient
+    """
+
+    userid: uuid.UUID = Field(required=False, primary_key=True, default=auto_uuid)
+    name: str
+    account: Optional[Account]
+    accounts: Optional[list[Account]]
+
+    def __str__(self) -> str:
+        return f"<{self.name}: {self.userid}>"
+
+class Message(BaseModel):
+    """
+    Message.
+    Base-class for Message blocks for Notify
+    TODO:
+     * template needs a factory function to find a jinja processor
+     *
+    """
+
+    name: str = Field(required=True, default=auto_uuid)
+    body: Union[str, dict] = Field(default=None)
+    content: str = Field(required=False, default="")
+    sent: datetime = Field(required=False, default=now)
+    template: Path
+
+class Attachment(BaseModel):
+    """Attachement.
+
+    an Attachment is any document attached to a message.
+    """
+
+    name: str = Field(required=True)
+    content: Any = None
+    content_type: str
+    type: str
+
+class BlockMessage(Message):
+    """
+    BlockMessage.
+    Class for Message Notifications
+    TODO:
+     * template needs a factory function to find a jinja processor
+     *
+    """
+
+    sender: Union[Actor, list[Actor]] = Field(required=False)
+    recipient: Union[Actor, list[Actor]] = Field(required=False)
+    content_type: Literal[
+        "text/plain",
+        "text/html",
+        "multipart/alternative",
+        "application/json"
+    ] = Field(default="text/plain")
+    attachments: list[Attachment] = Field(default_factory=list)
+    flags: list[str]
+
+def test_actor_valid():
+    a = Actor(
+        name="Alice",
+        account=Account(provider="prov", enabled=True)
+    )
+    assert a.name == "Alice"
+    assert a.account.provider == "prov"
+    # accounts can be None or list[Account]
+    a2 = Actor(
+        name="Bob",
+        accounts=[Account(provider="prov2")]
+    )
+    assert a2.accounts[0].provider == "prov2"
+
+def test_actor_invalid():
+    # Provide a wrong type for 'name'
+    actor = Actor(name={"user": 123})
+    print(actor.name, type(actor.name))
+
+if __name__ == "__main__":
+    test_actor_invalid()
+    print("test_notify.py: all tests passed")