Flat models structure generation

Author: bogdandm

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/models/__init__.py

@@ -1,4 +1,5 @@
-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
 
@@ -28,13 +29,36 @@ 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
+
+
+class PositionsDict(defaultdict):
+    INC = object()
+
+    def __init__(self, default_factory=int, **kwargs):
+        super().__init__(default_factory, **kwargs)
+
+    def update_position(self, key: str, value: Union[object, int]):
+        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:
@@ -91,6 +115,59 @@ 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(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

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.*