Filter import * from megasniff fields
This commit is contained in:
@@ -1,97 +1 @@
|
|||||||
# Copyright (C) 2025 Shevchenko A
|
from .inflator import SchemaInflatorGenerator
|
||||||
# SPDX-License-Identifier: LGPL-3.0-or-later
|
|
||||||
|
|
||||||
import importlib.resources
|
|
||||||
from collections.abc import Callable
|
|
||||||
from dataclasses import dataclass
|
|
||||||
from types import NoneType, UnionType
|
|
||||||
from typing import Optional, get_origin, get_args, Union, Annotated
|
|
||||||
|
|
||||||
import jinja2
|
|
||||||
|
|
||||||
from .utils import *
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class RenderData:
|
|
||||||
argname: str
|
|
||||||
constrs: list[tuple[str, bool]] # typecall / use lookup table
|
|
||||||
typename: str
|
|
||||||
is_union: bool
|
|
||||||
is_optional: bool
|
|
||||||
default_option: Optional[str]
|
|
||||||
|
|
||||||
|
|
||||||
class SchemaInflatorGenerator:
|
|
||||||
templateLoader: jinja2.BaseLoader
|
|
||||||
templateEnv: jinja2.Environment
|
|
||||||
template: jinja2.Template
|
|
||||||
|
|
||||||
def __init__(self,
|
|
||||||
loader: Optional[jinja2.BaseLoader] = None,
|
|
||||||
convertor_template: str = 'inflator.jinja2'):
|
|
||||||
if loader is None:
|
|
||||||
template_path = importlib.resources.files('megasniff.templates')
|
|
||||||
loader = jinja2.FileSystemLoader(str(template_path))
|
|
||||||
self.templateLoader = loader
|
|
||||||
self.templateEnv = jinja2.Environment(loader=self.templateLoader)
|
|
||||||
self.template = self.templateEnv.get_template(convertor_template)
|
|
||||||
|
|
||||||
def schema_to_generator(self,
|
|
||||||
schema: type,
|
|
||||||
*,
|
|
||||||
_base_lookup_table: Optional[dict[str, Any]] = None) -> Callable[[dict[str, Any]], Any]:
|
|
||||||
# Я это написал, оно пока работает, и я не собираюсь это упрощать, сорян
|
|
||||||
type_hints = get_kwargs_type_hints(schema)
|
|
||||||
render_data = []
|
|
||||||
lookup_table = _base_lookup_table or {}
|
|
||||||
|
|
||||||
if schema.__name__ not in lookup_table.keys():
|
|
||||||
lookup_table[schema.__name__] = None
|
|
||||||
|
|
||||||
for argname, argtype in type_hints.items():
|
|
||||||
if argname in {'return', 'self'}:
|
|
||||||
continue
|
|
||||||
|
|
||||||
has_default, default_option = get_field_default(schema, argname)
|
|
||||||
argtypes = argtype,
|
|
||||||
type_origin = get_origin(argtype)
|
|
||||||
|
|
||||||
if any(map(lambda x: type_origin is x, [Union, UnionType, Optional, Annotated])):
|
|
||||||
argtypes = get_args(argtype)
|
|
||||||
|
|
||||||
if NoneType in argtypes or None in argtypes:
|
|
||||||
argtypes = tuple(filter(lambda x: x is not None and x is not NoneType, argtypes))
|
|
||||||
has_default = True
|
|
||||||
|
|
||||||
out_argtypes: list[tuple[str, bool]] = []
|
|
||||||
|
|
||||||
for argt in argtypes:
|
|
||||||
is_builtin = is_builtin_type(argt)
|
|
||||||
if not is_builtin and argt is not schema:
|
|
||||||
if argt.__name__ not in lookup_table.keys():
|
|
||||||
# если случилась циклическая зависимость, мы не хотим бексконечную рекурсию
|
|
||||||
lookup_table[argt.__name__] = self.schema_to_generator(argt, _base_lookup_table=lookup_table)
|
|
||||||
if argt is schema:
|
|
||||||
out_argtypes.append(('inflate', True))
|
|
||||||
else:
|
|
||||||
out_argtypes.append((argt.__name__, is_builtin))
|
|
||||||
|
|
||||||
render_data.append(
|
|
||||||
RenderData(argname, out_argtypes, repr(argtype), len(argtypes) > 1, has_default, default_option))
|
|
||||||
|
|
||||||
convertor_functext = self.template.render(conversions=render_data)
|
|
||||||
convertor_functext = '\n'.join(list(filter(lambda x: len(x.strip()), convertor_functext.split('\n'))))
|
|
||||||
convertor_functext = convertor_functext.replace(', )', ')')
|
|
||||||
namespace = {
|
|
||||||
'_tgt_type': schema,
|
|
||||||
'_lookup_table': lookup_table
|
|
||||||
}
|
|
||||||
exec(convertor_functext, namespace)
|
|
||||||
|
|
||||||
# пихаем сгенеренный метод в табличку,
|
|
||||||
# ожидаем что она обновится во всех вложенных методах,
|
|
||||||
# разрешая циклические зависимости
|
|
||||||
lookup_table[schema.__name__] = namespace['inflate']
|
|
||||||
|
|
||||||
return namespace['inflate']
|
|
||||||
|
|||||||
97
src/megasniff/inflator.py
Normal file
97
src/megasniff/inflator.py
Normal file
@@ -0,0 +1,97 @@
|
|||||||
|
# Copyright (C) 2025 Shevchenko A
|
||||||
|
# SPDX-License-Identifier: LGPL-3.0-or-later
|
||||||
|
|
||||||
|
import importlib.resources
|
||||||
|
from collections.abc import Callable
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from types import NoneType, UnionType
|
||||||
|
from typing import Optional, get_origin, get_args, Union, Annotated
|
||||||
|
|
||||||
|
import jinja2
|
||||||
|
|
||||||
|
from .utils import *
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class RenderData:
|
||||||
|
argname: str
|
||||||
|
constrs: list[tuple[str, bool]] # typecall / use lookup table
|
||||||
|
typename: str
|
||||||
|
is_union: bool
|
||||||
|
is_optional: bool
|
||||||
|
default_option: Optional[str]
|
||||||
|
|
||||||
|
|
||||||
|
class SchemaInflatorGenerator:
|
||||||
|
templateLoader: jinja2.BaseLoader
|
||||||
|
templateEnv: jinja2.Environment
|
||||||
|
template: jinja2.Template
|
||||||
|
|
||||||
|
def __init__(self,
|
||||||
|
loader: Optional[jinja2.BaseLoader] = None,
|
||||||
|
convertor_template: str = 'inflator.jinja2'):
|
||||||
|
if loader is None:
|
||||||
|
template_path = importlib.resources.files('megasniff.templates')
|
||||||
|
loader = jinja2.FileSystemLoader(str(template_path))
|
||||||
|
self.templateLoader = loader
|
||||||
|
self.templateEnv = jinja2.Environment(loader=self.templateLoader)
|
||||||
|
self.template = self.templateEnv.get_template(convertor_template)
|
||||||
|
|
||||||
|
def schema_to_generator(self,
|
||||||
|
schema: type,
|
||||||
|
*,
|
||||||
|
_base_lookup_table: Optional[dict[str, Any]] = None) -> Callable[[dict[str, Any]], Any]:
|
||||||
|
# Я это написал, оно пока работает, и я не собираюсь это упрощать, сорян
|
||||||
|
type_hints = get_kwargs_type_hints(schema)
|
||||||
|
render_data = []
|
||||||
|
lookup_table = _base_lookup_table or {}
|
||||||
|
|
||||||
|
if schema.__name__ not in lookup_table.keys():
|
||||||
|
lookup_table[schema.__name__] = None
|
||||||
|
|
||||||
|
for argname, argtype in type_hints.items():
|
||||||
|
if argname in {'return', 'self'}:
|
||||||
|
continue
|
||||||
|
|
||||||
|
has_default, default_option = get_field_default(schema, argname)
|
||||||
|
argtypes = argtype,
|
||||||
|
type_origin = get_origin(argtype)
|
||||||
|
|
||||||
|
if any(map(lambda x: type_origin is x, [Union, UnionType, Optional, Annotated])):
|
||||||
|
argtypes = get_args(argtype)
|
||||||
|
|
||||||
|
if NoneType in argtypes or None in argtypes:
|
||||||
|
argtypes = tuple(filter(lambda x: x is not None and x is not NoneType, argtypes))
|
||||||
|
has_default = True
|
||||||
|
|
||||||
|
out_argtypes: list[tuple[str, bool]] = []
|
||||||
|
|
||||||
|
for argt in argtypes:
|
||||||
|
is_builtin = is_builtin_type(argt)
|
||||||
|
if not is_builtin and argt is not schema:
|
||||||
|
if argt.__name__ not in lookup_table.keys():
|
||||||
|
# если случилась циклическая зависимость, мы не хотим бексконечную рекурсию
|
||||||
|
lookup_table[argt.__name__] = self.schema_to_generator(argt, _base_lookup_table=lookup_table)
|
||||||
|
if argt is schema:
|
||||||
|
out_argtypes.append(('inflate', True))
|
||||||
|
else:
|
||||||
|
out_argtypes.append((argt.__name__, is_builtin))
|
||||||
|
|
||||||
|
render_data.append(
|
||||||
|
RenderData(argname, out_argtypes, repr(argtype), len(argtypes) > 1, has_default, default_option))
|
||||||
|
|
||||||
|
convertor_functext = self.template.render(conversions=render_data)
|
||||||
|
convertor_functext = '\n'.join(list(filter(lambda x: len(x.strip()), convertor_functext.split('\n'))))
|
||||||
|
convertor_functext = convertor_functext.replace(', )', ')')
|
||||||
|
namespace = {
|
||||||
|
'_tgt_type': schema,
|
||||||
|
'_lookup_table': lookup_table
|
||||||
|
}
|
||||||
|
exec(convertor_functext, namespace)
|
||||||
|
|
||||||
|
# пихаем сгенеренный метод в табличку,
|
||||||
|
# ожидаем что она обновится во всех вложенных методах,
|
||||||
|
# разрешая циклические зависимости
|
||||||
|
lookup_table[schema.__name__] = namespace['inflate']
|
||||||
|
|
||||||
|
return namespace['inflate']
|
||||||
Reference in New Issue
Block a user