Add flat types deflator tests

README.md

@@ -1,6 +1,6 @@
 # megasniff
 
-### Автоматическая валидация данных по схеме и сборка объекта в одном флаконе
+### Автоматическая валидация данных по схеме, сборка и разборка объекта в одном флаконе
 
 #### Как применять:
 
@@ -18,7 +18,8 @@ class SomeSchema1:
     c: SomeSchema2 | str | None
 
 
-class SomeSchema2(typing.TypedDict):
+@dataclasses.dataclass
+class SomeSchema2:
     field1: dict
     field2: float
     field3: typing.Optional[SomeSchema1]
@@ -28,12 +29,16 @@ class SomeSchema2(typing.TypedDict):
 import megasniff
 
 infl = megasniff.SchemaInflatorGenerator()
-fn = infl.schema_to_inflator(SomeSchema1)
+defl = megasniff.SchemaDeflatorGenerator()
+fn_in = infl.schema_to_inflator(SomeSchema1)
+fn_out = defl.schema_to_deflator(SomeSchema1)
 
 # 3. Проверяем что всё работает
 
-fn({'a': 1, 'b': 2, 'c': {'field1': {}, 'field2': '1.1', 'field3': None}})
+data = fn_in({'a': 1, 'b': 2, 'c': {'field1': {}, 'field2': '1.1', 'field3': None}})
 # SomeSchema1(a=1, b=2.0, c={'field1': {}, 'field2': 1.1, 'field3': None})
+fn_out(data)
+# {'a': 1, 'b': 2.0, 'c': {'field1': {}, 'field2': 1.1, 'field3': None}}
 
 ```
 
@@ -47,7 +52,11 @@ fn({'a': 1, 'b': 2, 'c': {'field1': {}, 'field2': '1.1', 'field3': None}})
 - не проверяет типы generic-словарей, кортежей (реализация ожидается)
 - пользовательские проверки типов должны быть реализованы через наследование и проверки в конструкторе
 - опциональный `strict-mode`: выключение приведения базовых типов
-- может генерировать кортежи верхнеуровневых объектов при наличии описания схемы (полезно при развертывании аргументов)
+- для inflation может генерировать кортежи верхнеуровневых объектов при наличии описания схемы (полезно при
+  развертывании аргументов)
+- `TypedDict` поддерживается только для inflation из-за сложностей выбора варианта при сборке `Union`-полей
+- для deflation поддерживается включение режима `explicit_casts`, приводящего типы к тем, которые указаны в
+  аннотациях (не распространяется на `Union`-типы, т.к. невозможно определить какой из них должен быть выбран)
 
 ----
 
@@ -88,6 +97,9 @@ class A:
 ```
 >>> {"a": [1, 1.1, "321"]}
 <<< A(a=[1, 1, 321])
+>>> A(a=[1, 1.1, "321"])
+<<< {"a": [1, 1.1, "321"]}  # explicit_casts=False
+<<< {"a": [1, 1, 321]}  # explicit_casts=True
 ```
 
 #### Strict-mode on:
@@ -101,11 +113,15 @@ class A:
 ```
 >>> {"a": [1, 1.1, "321"]}
 <<< FieldValidationException, т.к. 1.1 не является int
+>>> A(a=[1, 1.1, "321"])
+<<< FieldValidationException, т.к. 1.1 не является int
 ```
 
 ### Tuple unwrap
+
 ```
 fn = infl.schema_to_inflator(
         (('a', int), TupleSchemaItem(Optional[list[int]], key_name='b', has_default=True, default=None)))
 ```
+
 Создаёт `fn: (dict[str,Any]) -> tuple[int, Optional[list[int]]]: ...` (сигнатура остаётся `(dict[str,Any])->tuple`)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "megasniff"
-version = "0.2.3.post2"
+version = "0.2.6.post1"
 description = "Library for in-time codegened type validation"
 authors = [
     { name = "nikto_b", email = "niktob560@yandex.ru" }

src/megasniff/__main__.py

@@ -81,10 +81,11 @@ class ZSchema:
 
 def main_deflator():
     deflator = SchemaDeflatorGenerator(store_sources=True, explicit_casts=True, strict_mode=True)
-    fn = deflator.schema_to_deflator(DSchema)
-    print(getattr(fn, '__megasniff_sources__', '## No data'))
+    fn = deflator.schema_to_deflator(DSchema | int)
+
     # ret = fn(ZSchema(ZSchema(ZSchema(None, 42), 42), ZSchema(None, 42)))
-    ret = fn(DSchema({'a': 34}, {}, ASchema(1, 'a', None), ESchema([[['a'], ['b']]], ['b'])))
+    ret = fn(DSchema({'a': 34}, {}, ASchema(1, 'a', None), ESchema([[['a'], ['b']]], 'b')))
+    ret = fn(42)
     # assert ret['a'] == 1
     # assert ret['b'] == 1.1
     # assert ret['c'] == 'a'

src/megasniff/deflator.py

@@ -13,6 +13,10 @@ from typing import get_args, Union, Annotated, Sequence, TypeAliasType, \
 import jinja2
 
 from .utils import *
+import uuid
+from pathlib import Path
+import tempfile
+import importlib.util
 
 JsonObject: TypeAlias = Union[None, bool, int, float, str, list['JsonObject'], dict[str, 'JsonObject']]
 
@@ -99,7 +103,12 @@ def _flatten_type(t: type | TypeAliasType) -> tuple[type, Optional[str]]:
 
 def _schema_to_deflator_func(t: type | TypeAliasType) -> str:
     t, _ = _flatten_type(t)
-    return 'deflate_' + typename(t).replace('.', '_')
+    return ('deflate_' + typename(t)
+            .replace('.', '_')
+            .replace('[', '_of_')
+            .replace(']', '_of_')
+            .replace(',', '_and_')
+            .replace(' ', '_'))
 
 
 def _fallback_unwrapper(obj: Any) -> JsonObject:
@@ -129,6 +138,7 @@ class SchemaDeflatorGenerator:
     _store_sources: bool
     _strict_mode: bool
     _explicit_casts: bool
+    _out_directory: str | None
 
     def __init__(self,
                  loader: Optional[jinja2.BaseLoader] = None,
@@ -137,11 +147,13 @@ class SchemaDeflatorGenerator:
                  store_sources: bool = False,
                  *,
                  object_template_filename: str = 'deflator.jinja2',
+                 out_directory: str | None = None,
                  ):
 
         self._strict_mode = strict_mode
         self._store_sources = store_sources
         self._explicit_casts = explicit_casts
+        self._out_directory = out_directory
 
         if loader is None:
             template_path = importlib.resources.files('megasniff.templates')
@@ -155,16 +167,44 @@ class SchemaDeflatorGenerator:
                            schema: type,
                            strict_mode_override: Optional[bool] = None,
                            explicit_casts_override: Optional[bool] = None,
+                           ignore_directory: bool = False,
+                           out_directory_override: Optional[str] = None,
                            ) -> Callable[[Any], dict[str, Any]]:
         txt, namespace = self._schema_to_deflator(schema,
                                                   strict_mode_override=strict_mode_override,
                                                   explicit_casts_override=explicit_casts_override,
                                                   )
+
+        out_dir = self._out_directory
+        if out_directory_override:
+            out_dir = out_directory_override
+        if ignore_directory:
+            out_dir = None
+
         imports = ('from typing import Any\n'
                    'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
         txt = imports + '\n' + txt
-        exec(txt, namespace)
-        fn = namespace[_schema_to_deflator_func(schema)]
+
+        if out_dir is not None:
+            filename = f"{uuid.uuid4()}.py"
+            filepath = Path(out_dir) / filename
+            filepath.parent.mkdir(parents=True, exist_ok=True)
+
+            with open(filepath, 'w', encoding='utf-8') as f:
+                f.write(txt)
+
+            spec = importlib.util.spec_from_file_location("generated_module", filepath)
+            module = importlib.util.module_from_spec(spec)
+            module.__dict__.update(namespace)
+
+            spec.loader.exec_module(module)
+
+            fn_name = _schema_to_deflator_func(schema)
+            fn = getattr(module, fn_name)
+        else:
+            exec(txt, namespace)
+            fn = namespace[_schema_to_deflator_func(schema)]
+
         if self._store_sources:
             setattr(fn, '__megasniff_sources__', txt)
         return fn
@@ -244,7 +284,7 @@ class SchemaDeflatorGenerator:
 
                 ret_unw = ObjectUnwrapping(fields)
             else:
-                raise NotImplementedError()
+                raise NotImplementedError(f'type not implemented yet: {schema}')
 
         return ret_unw, field_rename, set(_visited_types) | ongoing_types, recurcive_types
 

src/megasniff/inflator.py

@@ -73,6 +73,7 @@ class SchemaInflatorGenerator:
     tuple_template: jinja2.Template
     _store_sources: bool
     _strict_mode: bool
+    _out_directory: str | None
 
     def __init__(self,
                  loader: Optional[jinja2.BaseLoader] = None,
@@ -81,10 +82,12 @@ class SchemaInflatorGenerator:
                  *,
                  object_template_filename: str = 'inflator.jinja2',
                  tuple_template_filename: str = 'inflator_tuple.jinja2',
+                 out_directory: str | None = None,
                  ):
 
         self._strict_mode = strict_mode
         self._store_sources = store_sources
+        self._out_directory = out_directory
 
         if loader is None:
             template_path = importlib.resources.files('megasniff.templates')
@@ -98,7 +101,9 @@ class SchemaInflatorGenerator:
     def schema_to_inflator(self,
                            schema: type | Sequence[TupleSchemaItem | tuple[str, type]] | OrderedDict[str, type],
                            strict_mode_override: Optional[bool] = None,
-                           from_type_override: Optional[type | TypeAliasType] = None
+                           from_type_override: Optional[type | TypeAliasType] = None,
+                           ignore_directory: bool = False,
+                           out_directory_override: Optional[str] = None,
                            ) -> Callable[[dict[str, Any]], Any]:
         if from_type_override is not None and '__getitem__' not in dir(from_type_override):
             raise RuntimeError('from_type_override must provide __getitem__')
@@ -107,11 +112,36 @@ class SchemaInflatorGenerator:
                                                   strict_mode_override=strict_mode_override,
                                                   from_type_override=from_type_override,
                                                   )
+        out_dir = self._out_directory
+        if out_directory_override:
+            out_dir = out_directory_override
+        if ignore_directory:
+            out_dir = None
+
         imports = ('from typing import Any\n'
                    'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
         txt = imports + '\n' + txt
-        exec(txt, namespace)
-        fn = namespace['inflate']
+
+        if out_dir is not None:
+            filename = f"{uuid.uuid4()}.py"
+            filepath = Path(out_dir) / filename
+            filepath.parent.mkdir(parents=True, exist_ok=True)
+
+            with open(filepath, 'w', encoding='utf-8') as f:
+                f.write(txt)
+
+            spec = importlib.util.spec_from_file_location("generated_module", filepath)
+            module = importlib.util.module_from_spec(spec)
+            module.__dict__.update(namespace)
+
+            spec.loader.exec_module(module)
+
+            fn_name = _schema_to_deflator_func(schema)
+            fn = getattr(module, fn_name)
+        else:
+            exec(txt, namespace)
+            fn = namespace['inflate']
+
         if self._store_sources:
             setattr(fn, '__megasniff_sources__', txt)
         return fn

src/megasniff/templates/deflator.jinja2

@@ -20,7 +20,7 @@ if not isinstance({{from_container}}, dict):
 {% endif %}
 for k_{{hashname(unwrapping)}}, v_{{hashname(unwrapping)}} in {{from_container}}.items():
 {{ render_unwrap(unwrapping.key_unwrap, 'k_' + hashname(unwrapping), 'k_' + hashname(unwrapping)) | indent(4) }}
-{{ render_unwrap(unwrapping.value_unwrap, into_container + '[v_' + hashname(unwrapping) + ']', 'v_' + hashname(unwrapping)) | indent(4) }}
+{{ render_unwrap(unwrapping.value_unwrap, 'v_' + hashname(unwrapping), into_container + '[k_' + hashname(unwrapping) + ']') | indent(4) }}
 {%- endset %}
 {{out}}
 {%- endmacro %}

src/megasniff/utils.py

@@ -68,13 +68,25 @@ def is_builtin_type(tp: type) -> bool:
 
 
 def typename(tp: type) -> str:
+    ret = ''
     if get_origin(tp) is None and hasattr(tp, '__name__'):
-        return tp.__name__
-    return str(tp)
+        ret = tp.__name__
+    else:
+        ret = str(tp)
+    ret = (ret
+           .replace('.', '_')
+           .replace('[', '_of_')
+           .replace(']', '_of_')
+           .replace(',', '_and_')
+           .replace(' ', '_')
+           .replace('\'', '')
+           .replace('<', '')
+           .replace('>', ''))
+    return ret
 
 
 def is_class_definition(obj):
-    return isinstance(obj, type) or inspect.isclass(obj)
+    return (isinstance(obj, type) or inspect.isclass(obj))
 
 
 def hashname(obj) -> str:

tests/test_basic_deflator.py

@@ -44,6 +44,20 @@ def test_unions():
     assert a['a'] == '42a'
 
 
+def test_dict_body():
+    @dataclass
+    class A:
+        a: dict[str, float]
+
+    defl = SchemaDeflatorGenerator()
+    fn = defl.schema_to_deflator(A)
+
+    a = fn(A({'1': 1.1, '2': 2.2}))
+    print(a)
+    assert a['a']['1'] == 1.1
+    assert a['a']['2'] == 2.2
+
+
 @dataclass
 class CircA:
     b: CircB

tests/test_flat_types_deflator.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Optional
+
+import pytest
+
+from megasniff.deflator import SchemaDeflatorGenerator
+from src.megasniff import SchemaInflatorGenerator
+
+
+def test_str_deflator():
+    defl = SchemaDeflatorGenerator()
+    fn = defl.schema_to_deflator(str, explicit_casts_override=True)
+    a = fn('asdf')
+
+    assert a == 'asdf'
+    a = fn(1234)
+
+    assert a == '1234'
+
+    fn1 = defl.schema_to_deflator(str, strict_mode_override=True)
+    with pytest.raises(Exception):
+        fn1(1234)
+
+
+def test_int_deflator():
+    defl = SchemaDeflatorGenerator()
+    fn = defl.schema_to_deflator(int, explicit_casts_override=True)
+    a = fn(1234)
+
+    assert a == 1234
+    a = fn('1234')
+
+    assert a == 1234
+
+    fn1 = defl.schema_to_deflator(int, strict_mode_override=True)
+    with pytest.raises(Exception):
+        fn1('1234')
+
+
+def test_float_deflator():
+    defl = SchemaDeflatorGenerator()
+    fn = defl.schema_to_deflator(float, explicit_casts_override=True)
+    a = fn(1234.1)
+
+    assert a == 1234.1
+    a = fn('1234')
+
+    assert a == 1234.0
+
+    fn1 = defl.schema_to_deflator(float, strict_mode_override=True)
+    with pytest.raises(Exception):
+        fn1(1234)