Experimental support of a file-based inflators/deflators

pyproject.toml

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

src/megasniff/deflator.py

@@ -134,6 +134,7 @@ class SchemaDeflatorGenerator:
     _store_sources: bool
     _strict_mode: bool
     _explicit_casts: bool
+    _out_directory: str | None
 
     def __init__(self,
                  loader: Optional[jinja2.BaseLoader] = None,
@@ -142,11 +143,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')
@@ -160,16 +163,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

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