String-convert "string" types in case they're not actually strings (UUIDs). Split out formats from types to allow for slightly more granular serializing.

Author: explorigin

pydanticrud/backends/dynamodb.py

@@ -1,8 +1,7 @@
-from typing import Optional, Set
+from typing import Optional, Set, Union
 import logging
 import json
 from datetime import datetime
-from base64 import b64encode, b64decode
 
 import boto3
 from boto3.dynamodb.conditions import Key, Attr
@@ -80,20 +79,16 @@ def _to_epoch_float(dt):
 
 SERIALIZE_MAP = {
     "number": str,  # float or decimal
-    "string": lambda d: d.isoformat() if isinstance(d, datetime) else d,  # string, datetime
+    "string": str,
+    "string:date-time": lambda d: d.isoformat(),
     "boolean": lambda d: 1 if d else 0,
     "object": json.dumps,
     "array": json.dumps,
 }
 
 
-def do_nothing(x):
-    return x
-
-
 DESERIALIZE_MAP = {
     "number": float,
-    "string": do_nothing,
     "boolean": bool,
     "object": json.loads,
     "array": json.loads,
@@ -121,27 +116,44 @@ def __init__(self, schema):
         self.properties = schema["properties"]
         self.definitions = schema.get("definitions")
 
-    def _get_type_possibilities(self, field_name) -> Set[str]:
+    def _get_type_possibilities(self, field_name) -> Set[tuple]:
         field_properties = self.properties[field_name]
 
         possible_types = []
         if "anyOf" in field_properties:
-            possible_types.extend([r.get("$ref", r.get("type")) for r in field_properties["anyOf"]])
+            possible_types.extend([r.get("$ref", r) for r in field_properties["anyOf"]])
         else:
-            possible_types.append(field_properties.get("$ref", field_properties.get("type")))
-        return set([
-            (self.definitions[t.split('/')[-1]].get("type") if t.startswith('#/') else t)
+            possible_types.append(field_properties.get("$ref", field_properties))
+
+        def type_from_definition(definition_signature: Union[str, dict]) -> dict:
+            if isinstance(definition_signature, str):
+                t = definition_signature.split('/')[-1]
+                return self.definitions[t]
+            return definition_signature
+
+        type_dicts = [
+            type_from_definition(t)
             for t in possible_types
+        ]
+
+        return set([
+            (t['type'], t.get('format', ''))
+            for t in type_dicts
         ])
 
     def _serialize_field(self, field_name, value):
         field_types = self._get_type_possibilities(field_name)
         if value is not None:
             for t in field_types:
                 try:
-                    return SERIALIZE_MAP[t](value)
+                    type_signature = ":".join(t).rstrip(':')
+                    try:
+                        return SERIALIZE_MAP[type_signature](value)
+                    except KeyError:
+                        return SERIALIZE_MAP[t[0]](value)
                 except (ValueError, TypeError, KeyError):
                     pass
+
         return value
 
     def serialize_record(self, data_dict) -> dict:
@@ -158,9 +170,14 @@ def _deserialize_field(self, field_name, value):
         if value is not None:
             for t in field_types:
                 try:
-                    return DESERIALIZE_MAP[t](value)
+                    type_signature = ":".join(t).rstrip(':')
+                    try:
+                        return DESERIALIZE_MAP[type_signature](value)
+                    except KeyError:
+                        return DESERIALIZE_MAP[t[0]](value)
                 except (ValueError, TypeError, KeyError):
                     pass
+
         return value
 
     def deserialize_record(self, data_dict) -> dict:

tests/test_dynamodb.py

@@ -1,9 +1,7 @@
 from typing import Dict, List, Optional, Union
-from base64 import b64decode, b64encode
 from decimal import Decimal
 from datetime import datetime
-import json
-from uuid import uuid4
+from uuid import uuid4, UUID
 import random
 
 import docker
@@ -32,6 +30,7 @@ class SimpleKeyModel(BaseModel):
     enabled: bool
     data: Dict[int, int] = None
     items: List[int]
+    hash: UUID
 
     class Config:
         title = "ModelTitle123"
@@ -122,6 +121,7 @@ def simple_model_data_generator(**kwargs):
         enabled=random.choice((True, False)),
         data={random.randint(0, 1000): random.randint(0, 1000)},
         items=[random.randint(0, 100000), random.randint(0, 100000), random.randint(0, 100000)],
+        hash=uuid4()
     )
     data.update(kwargs)
     return data