Compare commits

...

2 Commits

Author SHA1 Message Date
4068af462b Experimental support of a file-based inflators/deflators 2025-10-16 21:01:02 +03:00
d4d7a68d7a Fix dict body validation 2025-10-16 20:51:19 +03:00
4 changed files with 68 additions and 7 deletions

View File

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

View File

@@ -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

View File

@@ -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

View File

@@ -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 + '[v_' + hashname(unwrapping) + ']') | indent(4) }}
{%- endset %}
{{out}}
{%- endmacro %}