add JSON -> LarkElement deserializer;

batch of other changes

Author: kkozik-amplify

hcl2/const.py

@@ -2,3 +2,4 @@
 
 START_LINE_KEY = "__start_line__"
 END_LINE_KEY = "__end_line__"
+IS_BLOCK = "__is_block__"

hcl2/rule_transformer/deserializer.py

@@ -1,16 +1,68 @@
 import json
+from functools import lru_cache
 from typing import Any, TextIO, List
 
+from regex import regex
+
+from hcl2 import parses
+from hcl2.const import IS_BLOCK
 from hcl2.rule_transformer.rules.abstract import LarkElement, LarkRule
+from hcl2.rule_transformer.rules.base import (
+    BlockRule,
+    AttributeRule,
+    BodyRule,
+    StartRule,
+)
+from hcl2.rule_transformer.rules.containers import (
+    TupleRule,
+    ObjectRule,
+    ObjectElemRule,
+    ObjectElemKeyExpressionRule,
+    ObjectElemKeyDotAccessor,
+    ObjectElemKeyRule,
+)
+from hcl2.rule_transformer.rules.expressions import ExprTermRule
+from hcl2.rule_transformer.rules.literal_rules import (
+    IdentifierRule,
+    IntLitRule,
+    FloatLitRule,
+)
+from hcl2.rule_transformer.rules.strings import (
+    StringRule,
+    InterpolationRule,
+    StringPartRule,
+)
+from hcl2.rule_transformer.rules.tokens import (
+    NAME,
+    EQ,
+    DBLQUOTE,
+    STRING_CHARS,
+    ESCAPED_INTERPOLATION,
+    INTERP_START,
+    RBRACE,
+    IntLiteral,
+    FloatLiteral,
+    RSQB,
+    LSQB,
+    COMMA,
+    DOT,
+    LBRACE,
+)
+from hcl2.rule_transformer.transformer import RuleTransformer
 from hcl2.rule_transformer.utils import DeserializationOptions
 
 
 class Deserializer:
     def __init__(self, options=DeserializationOptions()):
         self.options = options
 
+    @property
+    @lru_cache
+    def _transformer(self) -> RuleTransformer:
+        return RuleTransformer()
+
     def load_python(self, value: Any) -> LarkElement:
-        pass
+        return StartRule([self._deserialize(value)])
 
     def loads(self, value: str) -> LarkElement:
         return self.load_python(json.loads(value))
@@ -19,13 +71,209 @@ def load(self, file: TextIO) -> LarkElement:
         return self.loads(file.read())
 
     def _deserialize(self, value: Any) -> LarkElement:
-        pass
+        if isinstance(value, dict):
+            if self._contains_block_marker(value):
+                elements = self._deserialize_block_elements(value)
+                return BodyRule(elements)
+
+            return self._deserialize_object(value)
+
+        if isinstance(value, list):
+            return self._deserialize_list(value)
+
+        return self._deserialize_text(value)
+
+    def _deserialize_block_elements(self, value: dict) -> List[LarkRule]:
+        children = []
+
+        for key, value in value.items():
+            if self._is_block(value):
+                # this value is a list of blocks, iterate over each block and deserialize them
+                for block in value:
+                    children.append(self._deserialize_block(key, block))
+            else:
+
+                # otherwise it's just an attribute
+                if key != IS_BLOCK:
+                    children.append(self._deserialize_attribute(key, value))
+
+        return children
+
+    def _deserialize_text(self, value) -> LarkRule:
+        try:
+            int_val = int(value)
+            return IntLitRule([IntLiteral(int_val)])
+        except ValueError:
+            pass
+
+        try:
+            float_val = float(value)
+            return FloatLitRule([FloatLiteral(float_val)])
+        except ValueError:
+            pass
+
+        if isinstance(value, str):
+            if value.startswith('"') and value.endswith('"'):
+                return self._deserialize_string(value)
+
+            if self._is_expression(value):
+                return self._deserialize_expression(value)
+
+            return self._deserialize_identifier(value)
+
+        elif isinstance(value, bool):
+            return self._deserialize_identifier(str(value).lower())
+
+        return self._deserialize_identifier(str(value))
+
+    def _deserialize_identifier(self, value: str) -> IdentifierRule:
+        return IdentifierRule([NAME(value)])
+
+    def _deserialize_string(self, value: str) -> StringRule:
+        result = []
+
+        pattern = regex.compile(r"(\${1,2}\{(?:[^{}]|(?R))*\})")
+        parts = [part for part in pattern.split(value) if part != ""]
+        # e.g. 'aaa$${bbb}ccc${"ddd-${eee}"}' -> ['aaa', '$${bbb}', 'ccc', '${"ddd-${eee}"}']
+        # 'aa-${"bb-${"cc-${"dd-${5 + 5}"}"}"}' -> ['aa-', '${"bb-${"cc-${"dd-${5 + 5}"}"}"}']
+
+        for part in parts:
+            if part == '"':
+                continue
+
+            if part.startswith('"'):
+                part = part[1:]
+            if part.endswith('"'):
+                part = part[:-1]
+
+            e = self._deserialize_string_part(part)
+            result.append(e)
+
+        return StringRule([DBLQUOTE(), *result, DBLQUOTE()])
+
+    def _deserialize_string_part(self, value: str) -> StringPartRule:
+        if value.startswith("$${") and value.endswith("}"):
+            return StringPartRule([ESCAPED_INTERPOLATION(value)])
+
+        if value.startswith("${") and value.endswith("}"):
+            return StringPartRule(
+                [
+                    InterpolationRule(
+                        [INTERP_START(), self._deserialize_expression(value), RBRACE()]
+                    )
+                ]
+            )
+
+        return StringPartRule([STRING_CHARS(value)])
+
+    def _deserialize_expression(self, value: str) -> ExprTermRule:
+        """Deserialize an expression string into an ExprTermRule."""
+        # instead of processing expression manually and trying to recognize what kind of expression it is,
+        #   turn it into HCL2 code and parse it with lark:
+
+        # unwrap from ${ and }
+        value = value[2:-1]
+        # create HCL2 snippet
+        value = f"temp = {value}"
+        # parse the above
+        parsed_tree = parses(value)
+        # transform parsed tree into LarkElement tree
+        rules_tree = self._transformer.transform(parsed_tree)
+        # extract expression from the tree
+        return rules_tree.body.children[0].expression
+
+    def _deserialize_block(self, first_label: str, value: dict) -> BlockRule:
+        """Deserialize a block by extracting labels and body"""
+        labels = [first_label]
+        body = value
+
+        # Keep peeling off single-key layers until we hit the body (dict with IS_BLOCK)
+        while isinstance(body, dict) and not body.get(IS_BLOCK):
+            non_block_keys = [k for k in body.keys() if k != IS_BLOCK]
+            if len(non_block_keys) == 1:
+                # This is another label level
+                label = non_block_keys[0]
+                labels.append(label)
+                body = body[label]
+            else:
+                # Multiple keys = this is the body
+                break
+
+        return BlockRule(
+            [*[self._deserialize(label) for label in labels], self._deserialize(body)]
+        )
+
+    def _deserialize_attribute(self, name: str, value: Any) -> AttributeRule:
+        children = [
+            self._deserialize_identifier(name),
+            EQ(),
+            ExprTermRule([self._deserialize(value)]),
+        ]
+        return AttributeRule(children)
+
+    def _deserialize_list(self, value: List) -> TupleRule:
+        children = []
+        for element in value:
+            deserialized = self._deserialize(element)
+            if not isinstance(deserialized, ExprTermRule):
+                # whatever an element of the list is, it has to be nested inside ExprTermRule
+                deserialized = ExprTermRule([deserialized])
+            children.append(deserialized)
+            children.append(COMMA())
+
+        return TupleRule([LSQB(), *children, RSQB()])
+
+    def _deserialize_object(self, value: dict) -> ObjectRule:
+        children = []
+        for key, value in value.items():
+            children.append(self._deserialize_object_elem(key, value))
+        return ObjectRule([LBRACE(), *children, RBRACE()])
+
+    def _deserialize_object_elem(self, key: str, value: Any) -> ObjectElemRule:
+        if self._is_expression(key):
+            key = ObjectElemKeyExpressionRule([self._deserialize_expression(key)])
+        elif "." in key:
+            parts = key.split(".")
+            children = []
+            for part in parts:
+                children.append(self._deserialize_identifier(part))
+                children.append(DOT())
+            key = ObjectElemKeyDotAccessor(children[:-1])  # without the last comma
+        else:
+            key = self._deserialize_text(key)
+
+        return ObjectElemRule(
+            [
+                ObjectElemKeyRule([key]),
+                EQ(),
+                ExprTermRule([self._deserialize_text(value)]),
+            ]
+        )
+
+    def _is_expression(self, value: str) -> bool:
+        return value.startswith("${") and value.endswith("}")
+
+    def _is_block(self, value: Any) -> bool:
+        """Simple check: if it's a list containing dicts with IS_BLOCK markers"""
+        if not isinstance(value, list) or len(value) == 0:
+            return False
 
-    def _deserialize_dict(self, value: dict) -> LarkRule:
-        pass
+        # Check if any item in the list has IS_BLOCK marker (directly or nested)
+        for item in value:
+            if isinstance(item, dict) and self._contains_block_marker(item):
+                return True
 
-    def _deserialize_list(self, value: List) -> LarkRule:
-        pass
+        return False
 
-    def _deserialize_expression(self, value: str) -> LarkRule:
-        pass
+    def _contains_block_marker(self, obj: dict) -> bool:
+        """Recursively check if a dict contains IS_BLOCK marker anywhere"""
+        if obj.get(IS_BLOCK):
+            return True
+        for value in obj.values():
+            if isinstance(value, dict) and self._contains_block_marker(value):
+                return True
+            if isinstance(value, list):
+                for element in value:
+                    if self._contains_block_marker(element):
+                        return True
+        return False

hcl2/rule_transformer/rules/abstract.py

@@ -36,7 +36,7 @@ def serialize(
 
 
 class LarkToken(LarkElement, ABC):
-    def __init__(self, value: Union[str, int]):
+    def __init__(self, value: Union[str, int, float]):
         self._value = value
         super().__init__()
 
@@ -100,7 +100,6 @@ def __init__(self, children: List[LarkElement], meta: Optional[Meta] = None):
 
         for index, child in enumerate(children):
             if child is not None:
-                print(child)
                 child.set_index(index)
                 child.set_parent(self)
 

hcl2/rule_transformer/rules/base.py

@@ -3,9 +3,11 @@
 
 from lark.tree import Meta
 
-from hcl2.dict_transformer import START_LINE, END_LINE
+from hcl2.const import IS_BLOCK
 from hcl2.rule_transformer.rules.abstract import LarkRule, LarkToken
 from hcl2.rule_transformer.rules.expressions import ExpressionRule
+from hcl2.rule_transformer.rules.literal_rules import IdentifierRule
+from hcl2.rule_transformer.rules.strings import StringRule
 from hcl2.rule_transformer.rules.tokens import NAME, EQ
 
 from hcl2.rule_transformer.rules.whitespace import NewLineOrCommentRule
@@ -42,7 +44,7 @@ class BodyRule(LarkRule):
     _children: List[
         Union[
             NewLineOrCommentRule,
-            # AttributeRule,
+            AttributeRule,
             "BlockRule",
         ]
     ]
@@ -58,6 +60,7 @@ def serialize(
         attributes: List[AttributeRule] = []
         comments = []
         inline_comments = []
+
         for child in self._children:
 
             if isinstance(child, BlockRule):
@@ -116,7 +119,11 @@ def serialize(
 
 class BlockRule(LarkRule):
 
-    _children: Tuple[BodyRule]
+    _children: Tuple[
+        IdentifierRule,
+        Optional[Union[IdentifierRule, StringRule]],
+        BodyRule,
+    ]
 
     def __init__(self, children, meta: Optional[Meta] = None):
         super().__init__(children, meta)
@@ -141,15 +148,11 @@ def serialize(
         self, options=SerializationOptions(), context=SerializationContext()
     ) -> Any:
         result = self._body.serialize(options)
+        if options.explicit_blocks:
+            result.update({IS_BLOCK: True})
+
         labels = self._labels
         for label in reversed(labels[1:]):
             result = {label.serialize(options): result}
 
-        result.update(
-            {
-                START_LINE: self._meta.line,
-                END_LINE: self._meta.end_line,
-            }
-        )
-
         return result

hcl2/rule_transformer/rules/functions.py

@@ -78,7 +78,7 @@ def arguments(self) -> Optional[ArgumentsRule]:
 
     def serialize(self, options = SerializationOptions(), context = SerializationContext()) -> Any:
         result = (
-            f"{":".join(identifier.serialize(options, context) for identifier in self.identifiers)}"
+            f"{"::".join(identifier.serialize(options, context) for identifier in self.identifiers)}"
             f"({self.arguments.serialize(options, context) if self.arguments else ""})"
         )
         if not context.inside_dollar_string:

hcl2/rule_transformer/rules/literal_rules.py

@@ -43,12 +43,6 @@ def lark_name() -> str:
         return "float_lit"
 
 
-class StringPartRule(TokenRule):
-    @staticmethod
-    def lark_name() -> str:
-        return "string_part"
-
-
 class BinaryOperatorRule(TokenRule):
     @staticmethod
     def lark_name() -> str: