Flat models structure generation

json_to_models/cli.py

@@ -12,7 +12,7 @@
 import json_to_models
 from json_to_models.dynamic_typing import ModelMeta, register_datetime_classes
 from json_to_models.generator import MetadataGenerator
-from json_to_models.models import ModelsStructureType, compose_models
+from json_to_models.models import ModelsStructureType, compose_models, compose_models_flat
 from json_to_models.models.attr import AttrsModelCodeGenerator
 from json_to_models.models.base import GenericModelCodeGenerator, generate_code
 from json_to_models.models.dataclasses import DataclassModelCodeGenerator
@@ -34,8 +34,7 @@ class Cli:
 
     STRUCTURE_FN_MAPPING: Dict[str, STRUCTURE_FN_TYPE] = {
         "nested": compose_models,
-        # TODO: vvvvvvvvvvvv
-        "flat": lambda *args, **kwargs: None
+        "flat": compose_models_flat
     }
 
     MODEL_GENERATOR_MAPPING: Dict[str, Type[GenericModelCodeGenerator]] = {
@@ -187,8 +186,8 @@ def set_args(self, merge_policy: List[Union[List[str], str]],
 
         self.initialize = True
 
-    @staticmethod
-    def _create_argparser() -> argparse.ArgumentParser:
+    @classmethod
+    def _create_argparser(cls) -> argparse.ArgumentParser:
         """
         ArgParser factory
         """
@@ -255,13 +254,13 @@ def _create_argparser() -> argparse.ArgumentParser:
         parser.add_argument(
             "-s", "--structure",
             default="nested",
-            choices=["nested", "flat"],
+            choices=list(cls.STRUCTURE_FN_MAPPING.keys()),
             help="Models composition style. By default nested models become nested Python classes.\n\n"
         )
         parser.add_argument(
             "-f", "--framework",
             default="base",
-            choices=["base", "attrs", "dataclasses", "custom"],
+            choices=list(cls.MODEL_GENERATOR_MAPPING.keys()) + ["custom"],
             help="Model framework for which python code is generated.\n"
                  "'base' (default) mean no framework so code will be generated without any decorators\n"
                  "and additional meta-data.\n"

json_to_models/dynamic_typing/models_meta.py

@@ -23,6 +23,9 @@ def __init__(self, t: MetaData, index, _original_fields=None):
     def __str__(self):
         return f"Model#{self.index}" + ("-" + self._name if self._name else "")
 
+    def __repr__(self):
+        return f"<{self}>"
+
     def __eq__(self, other):
         if isinstance(other, dict):
             return self.type == other

json_to_models/generator.py

@@ -9,7 +9,7 @@
                              StringSerializable, StringSerializableRegistry, Unknown, registry)
 
 keywords_set = set(keyword.kwlist)
-
+_static_types = {float, bool, int}
 
 class MetadataGenerator:
     CONVERTER_TYPE = Optional[Callable[[str], Any]]
@@ -61,15 +61,12 @@ def _detect_type(self, value, convert_dict=True) -> MetaData:
         Converts json value to metadata
         """
         # Simple types
-        if isinstance(value, float):
-            return float
-        elif isinstance(value, bool):
-            return bool
-        elif isinstance(value, int):
-            return int
+        t = type(value)
+        if t in _static_types:
+            return t
 
         # List trying to yield nested type
-        elif isinstance(value, list):
+        elif t is list:
             if value:
                 types = [self._detect_type(item) for item in value]
                 if len(types) > 1:

json_to_models/models/__init__.py

@@ -1,8 +1,11 @@
-from typing import Dict, Generic, Iterable, List, Set, Tuple, TypeVar
+from collections import defaultdict
+from typing import Dict, Generic, Iterable, List, Set, Tuple, TypeVar, Union
 
 from ..dynamic_typing import DOptional, ModelMeta, ModelPtr
 
 Index = str
+ModelsStructureType = Tuple[List[dict], Dict[ModelMeta, ModelMeta]]
+
 T = TypeVar('T')
 
 
@@ -26,40 +29,44 @@ def insert_before(self, value: T, *before: T):
             raise ValueError
         pos = min(ix)
         self.insert(pos, value)
+        return pos
 
     def insert_after(self, value: T, *after: T):
         ix = self._safe_indexes(*after)
         if not ix:
             raise ValueError
-        pos = max(ix)
-        self.insert(pos + 1, value)
-
+        pos = max(ix) + 1
+        self.insert(pos, value)
+        return pos
 
-def filter_pointers(model: ModelMeta) -> Iterable[ModelPtr]:
-    """
-    Return iterator over pointers with not None parent
-    """
-    return (ptr for ptr in model.pointers if ptr.parent)
 
+class PositionsDict(defaultdict):
+    # Dict contains mapping Index -> position, where position is list index to insert nested element of Index
+    INC = object()
 
-def extract_root(model: ModelMeta) -> Set[Index]:
-    """
-    Return set of indexes of root models that are use given ``model`` directly or through another nested model.
-    """
-    seen: Set[Index] = set()
-    nodes: List[ModelPtr] = list(filter_pointers(model))
-    roots: Set[Index] = set()
-    while nodes:
-        node = nodes.pop()
-        seen.add(node.type.index)
-        filtered = list(filter_pointers(node.parent))
-        nodes.extend(ptr for ptr in filtered if ptr.type.index not in seen)
-        if not filtered:
-            roots.add(node.parent.index)
-    return roots
+    def __init__(self, default_factory=int, **kwargs):
+        super().__init__(default_factory, **kwargs)
 
+    def update_position(self, key: str, value: Union[object, int]):
+        """
+        Shift all elements which are placed after updated one
 
-ModelsStructureType = Tuple[List[dict], Dict[ModelMeta, ModelMeta]]
+        :param key: Index or "root"
+        :param value: Could be position or PositionsDict.INC to perform quick increment (x+=1)
+        :return:
+        """
+        if value is self.INC:
+            value = self[key] + 1
+        if key in self:
+            old_value = self[key]
+            delta = value - old_value
+        else:
+            old_value = value
+            delta = 1
+        for k, v in self.items():
+            if k != key and v >= old_value:
+                self[k] += delta
+        self[key] = value
 
 
 def compose_models(models_map: Dict[str, ModelMeta]) -> ModelsStructureType:
@@ -116,6 +123,84 @@ def compose_models(models_map: Dict[str, ModelMeta]) -> ModelsStructureType:
     return root_models, path_injections
 
 
+def compose_models_flat(models_map: Dict[Index, ModelMeta]) -> ModelsStructureType:
+    """
+    Generate flat sorted (by nesting level, ASC) models structure for internal usage.
+
+    :param models_map: Mapping (model index -> model meta instance).
+    :return: List of root models data, Map(child model -> root model) for absolute ref generation
+    """
+    root_models = ListEx()
+    positions: PositionsDict[Index, int] = PositionsDict()
+    top_level_models: Set[Index] = set()
+    structure_hash_table: Dict[Index, dict] = {
+        key: {
+            "model": model,
+            "nested": ListEx(),
+            "roots": list(extract_root(model)),  # Indexes of root level models
+        } for key, model in models_map.items()
+    }
+
+    for key, model in models_map.items():
+        pointers = list(filter_pointers(model))
+        has_root_pointers = len(pointers) != len(model.pointers)
+        if not pointers:
+            # Root level model
+            if not has_root_pointers:
+                raise Exception(f'Model {model.name} has no pointers')
+            root_models.insert(positions["root"], structure_hash_table[key])
+            top_level_models.add(key)
+            positions.update_position("root", PositionsDict.INC)
+        else:
+            parents = {ptr.parent.index for ptr in pointers}
+            struct = structure_hash_table[key]
+            # Model is using by other models
+            if has_root_pointers or len(parents) > 1 and len(struct["roots"]) >= 1:
+                # Model is using by different root models
+                if parents & top_level_models:
+                    parents.add("root")
+                parents_positions = {positions[parent_key] for parent_key in parents
+                                     if parent_key in positions}
+                parents_joined = "#".join(sorted(parents))
+                if parents_joined in positions:
+                    parents_positions.add(positions[parents_joined])
+                pos = max(parents_positions) if parents_positions else len(root_models)
+                positions.update_position(parents_joined, pos + 1)
+            else:
+                # Model is using by only one model
+                parent = next(iter(parents))
+                pos = positions.get(parent, len(root_models))
+                positions.update_position(parent, pos + 1)
+            positions.update_position(key, pos + 1)
+            root_models.insert(pos, struct)
+
+    return root_models, {}
+
+
+def filter_pointers(model: ModelMeta) -> Iterable[ModelPtr]:
+    """
+    Return iterator over pointers with not None parent
+    """
+    return (ptr for ptr in model.pointers if ptr.parent)
+
+
+def extract_root(model: ModelMeta) -> Set[Index]:
+    """
+    Return set of indexes of root models that are use given ``model`` directly or through another nested model.
+    """
+    seen: Set[Index] = set()
+    nodes: List[ModelPtr] = list(filter_pointers(model))
+    roots: Set[Index] = set()
+    while nodes:
+        node = nodes.pop()
+        seen.add(node.type.index)
+        filtered = list(filter_pointers(node.parent))
+        nodes.extend(ptr for ptr in filtered if ptr.type.index not in seen)
+        if not filtered:
+            roots.add(node.parent.index)
+    return roots
+
+
 def sort_fields(model_meta: ModelMeta) -> Tuple[List[str], List[str]]:
     """
     Split fields into required and optional groups

json_to_models/registry.py

@@ -2,6 +2,8 @@
 from itertools import chain, combinations
 from typing import Dict, List, Set, Tuple
 
+from ordered_set import OrderedSet
+
 from .dynamic_typing import BaseType, MetaData, ModelMeta, ModelPtr
 from .utils import Index, distinct_words
 
@@ -151,7 +153,7 @@ def merge_models(self, generator, strict=False) -> List[Tuple[ModelMeta, Set[Mod
         flag = True
         while flag:
             flag = False
-            new_groups: Set[Set[ModelMeta]] = set()
+            new_groups: OrderedSet[Set[ModelMeta]] = OrderedSet()
             for gr1, gr2 in combinations(groups, 2):
                 if gr1 & gr2:
                     old_len = len(new_groups)

requirements.txt

@@ -1,4 +1,5 @@
 python-dateutil>=2.7.*
 inflection>=0.3.*
 unidecode>=1.0.*
-Jinja2>=2.10.*
+Jinja2>=2.10.*
+ordered-set==3.*

setup.py

@@ -1,3 +1,4 @@
+import multiprocessing
 import sys
 
 from setuptools import find_packages, setup
@@ -9,6 +10,8 @@
     required = f.read().splitlines()
 URL = "https://github.com/bogdandm/json2python-models"
 
+CPU_N = multiprocessing.cpu_count()
+
 
 class PyTest(TestCommand):
     user_options = [("pytest-args=", "a", "Arguments to pass to pytest")]
@@ -20,8 +23,10 @@ def initialize_options(self):
     def run_tests(self):
         import shlex
         import pytest
-
-        errno = pytest.main(shlex.split(self.pytest_args + ' -m "not slow_http"'))
+        args = self.pytest_args
+        if CPU_N > 1 and "-n " not in args:
+            args += f" -n {CPU_N}"
+        errno = pytest.main(shlex.split(args))
         sys.exit(errno)
 
 
@@ -31,6 +36,7 @@ def run_tests(self):
     python_requires=">=3.7",
     url=URL,
     author="bogdandm (Bogdan Kalashnikov)",
+    author_email="bogdan.dm1995@yandex.ru",
     description="Python models (attrs, dataclasses or custom) generator from JSON data with typing module support",
     license="MIT",
     packages=find_packages(exclude=['test', 'testing_tools']),
@@ -39,9 +45,6 @@ def run_tests(self):
     },
     install_requires=required,
     cmdclass={"test": PyTest},
-    tests_require=["pytest", "requests", "attrs"],
-    project_urls={
-        'Source': URL
-    },
+    tests_require=["pytest", "pytest-xdist", "requests", "attrs"],
     data_files=[('', ['pytest.ini', '.coveragerc', 'LICENSE'])]
 )

test/test_cli/test_script.py

@@ -92,6 +92,14 @@ def test_script(command):
     print(stdout)
 
 
+@pytest.mark.parametrize("command", test_commands)
+def test_script_flat(command):
+    command += " -s flat"
+    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    stdout, stderr = _validate_result(proc)
+    print(stdout)
+
+
 @pytest.mark.parametrize("command", test_commands)
 def test_script_attrs(command):
     command += " -f attrs"