Compare commits
4 Commits
b724e3c5dd
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 89c4bcae90 | |||
| 9775bc2cc6 | |||
| 4068af462b | |||
| d4d7a68d7a |
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "megasniff"
|
name = "megasniff"
|
||||||
version = "0.2.5.post1"
|
version = "0.2.6.post1"
|
||||||
description = "Library for in-time codegened type validation"
|
description = "Library for in-time codegened type validation"
|
||||||
authors = [
|
authors = [
|
||||||
{ name = "nikto_b", email = "niktob560@yandex.ru" }
|
{ name = "nikto_b", email = "niktob560@yandex.ru" }
|
||||||
|
|||||||
@@ -13,6 +13,10 @@ from typing import get_args, Union, Annotated, Sequence, TypeAliasType, \
|
|||||||
import jinja2
|
import jinja2
|
||||||
|
|
||||||
from .utils import *
|
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']]
|
JsonObject: TypeAlias = Union[None, bool, int, float, str, list['JsonObject'], dict[str, 'JsonObject']]
|
||||||
|
|
||||||
@@ -134,6 +138,7 @@ class SchemaDeflatorGenerator:
|
|||||||
_store_sources: bool
|
_store_sources: bool
|
||||||
_strict_mode: bool
|
_strict_mode: bool
|
||||||
_explicit_casts: bool
|
_explicit_casts: bool
|
||||||
|
_out_directory: str | None
|
||||||
|
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
loader: Optional[jinja2.BaseLoader] = None,
|
loader: Optional[jinja2.BaseLoader] = None,
|
||||||
@@ -142,11 +147,13 @@ class SchemaDeflatorGenerator:
|
|||||||
store_sources: bool = False,
|
store_sources: bool = False,
|
||||||
*,
|
*,
|
||||||
object_template_filename: str = 'deflator.jinja2',
|
object_template_filename: str = 'deflator.jinja2',
|
||||||
|
out_directory: str | None = None,
|
||||||
):
|
):
|
||||||
|
|
||||||
self._strict_mode = strict_mode
|
self._strict_mode = strict_mode
|
||||||
self._store_sources = store_sources
|
self._store_sources = store_sources
|
||||||
self._explicit_casts = explicit_casts
|
self._explicit_casts = explicit_casts
|
||||||
|
self._out_directory = out_directory
|
||||||
|
|
||||||
if loader is None:
|
if loader is None:
|
||||||
template_path = importlib.resources.files('megasniff.templates')
|
template_path = importlib.resources.files('megasniff.templates')
|
||||||
@@ -160,16 +167,44 @@ class SchemaDeflatorGenerator:
|
|||||||
schema: type,
|
schema: type,
|
||||||
strict_mode_override: Optional[bool] = None,
|
strict_mode_override: Optional[bool] = None,
|
||||||
explicit_casts_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]]:
|
) -> Callable[[Any], dict[str, Any]]:
|
||||||
txt, namespace = self._schema_to_deflator(schema,
|
txt, namespace = self._schema_to_deflator(schema,
|
||||||
strict_mode_override=strict_mode_override,
|
strict_mode_override=strict_mode_override,
|
||||||
explicit_casts_override=explicit_casts_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'
|
imports = ('from typing import Any\n'
|
||||||
'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
|
'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
|
||||||
txt = imports + '\n' + txt
|
txt = imports + '\n' + txt
|
||||||
|
|
||||||
|
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)
|
exec(txt, namespace)
|
||||||
fn = namespace[_schema_to_deflator_func(schema)]
|
fn = namespace[_schema_to_deflator_func(schema)]
|
||||||
|
|
||||||
if self._store_sources:
|
if self._store_sources:
|
||||||
setattr(fn, '__megasniff_sources__', txt)
|
setattr(fn, '__megasniff_sources__', txt)
|
||||||
return fn
|
return fn
|
||||||
|
|||||||
@@ -73,6 +73,7 @@ class SchemaInflatorGenerator:
|
|||||||
tuple_template: jinja2.Template
|
tuple_template: jinja2.Template
|
||||||
_store_sources: bool
|
_store_sources: bool
|
||||||
_strict_mode: bool
|
_strict_mode: bool
|
||||||
|
_out_directory: str | None
|
||||||
|
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
loader: Optional[jinja2.BaseLoader] = None,
|
loader: Optional[jinja2.BaseLoader] = None,
|
||||||
@@ -81,10 +82,12 @@ class SchemaInflatorGenerator:
|
|||||||
*,
|
*,
|
||||||
object_template_filename: str = 'inflator.jinja2',
|
object_template_filename: str = 'inflator.jinja2',
|
||||||
tuple_template_filename: str = 'inflator_tuple.jinja2',
|
tuple_template_filename: str = 'inflator_tuple.jinja2',
|
||||||
|
out_directory: str | None = None,
|
||||||
):
|
):
|
||||||
|
|
||||||
self._strict_mode = strict_mode
|
self._strict_mode = strict_mode
|
||||||
self._store_sources = store_sources
|
self._store_sources = store_sources
|
||||||
|
self._out_directory = out_directory
|
||||||
|
|
||||||
if loader is None:
|
if loader is None:
|
||||||
template_path = importlib.resources.files('megasniff.templates')
|
template_path = importlib.resources.files('megasniff.templates')
|
||||||
@@ -98,7 +101,9 @@ class SchemaInflatorGenerator:
|
|||||||
def schema_to_inflator(self,
|
def schema_to_inflator(self,
|
||||||
schema: type | Sequence[TupleSchemaItem | tuple[str, type]] | OrderedDict[str, type],
|
schema: type | Sequence[TupleSchemaItem | tuple[str, type]] | OrderedDict[str, type],
|
||||||
strict_mode_override: Optional[bool] = None,
|
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]:
|
) -> Callable[[dict[str, Any]], Any]:
|
||||||
if from_type_override is not None and '__getitem__' not in dir(from_type_override):
|
if from_type_override is not None and '__getitem__' not in dir(from_type_override):
|
||||||
raise RuntimeError('from_type_override must provide __getitem__')
|
raise RuntimeError('from_type_override must provide __getitem__')
|
||||||
@@ -107,11 +112,36 @@ class SchemaInflatorGenerator:
|
|||||||
strict_mode_override=strict_mode_override,
|
strict_mode_override=strict_mode_override,
|
||||||
from_type_override=from_type_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'
|
imports = ('from typing import Any\n'
|
||||||
'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
|
'from megasniff.exceptions import MissingFieldException, FieldValidationException\n')
|
||||||
txt = imports + '\n' + txt
|
txt = imports + '\n' + txt
|
||||||
|
|
||||||
|
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)
|
exec(txt, namespace)
|
||||||
fn = namespace['inflate']
|
fn = namespace['inflate']
|
||||||
|
|
||||||
if self._store_sources:
|
if self._store_sources:
|
||||||
setattr(fn, '__megasniff_sources__', txt)
|
setattr(fn, '__megasniff_sources__', txt)
|
||||||
return fn
|
return fn
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ if not isinstance({{from_container}}, dict):
|
|||||||
{% endif %}
|
{% endif %}
|
||||||
for k_{{hashname(unwrapping)}}, v_{{hashname(unwrapping)}} in {{from_container}}.items():
|
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.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 %}
|
{%- endset %}
|
||||||
{{out}}
|
{{out}}
|
||||||
{%- endmacro %}
|
{%- endmacro %}
|
||||||
|
|||||||
@@ -44,6 +44,20 @@ def test_unions():
|
|||||||
assert a['a'] == '42a'
|
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
|
@dataclass
|
||||||
class CircA:
|
class CircA:
|
||||||
b: CircB
|
b: CircB
|
||||||
|
|||||||
53
tests/test_flat_types_deflator.py
Normal file
53
tests/test_flat_types_deflator.py
Normal file
@@ -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)
|
||||||
Reference in New Issue
Block a user