✨ v3.1.3 Fix integer pipes, add self syntax for symlinking. (#224)

# v3.1.3

- **Fix syncing pipes with integer datetimes.**  
  An issue with dtype detection has been fixed, improving performance for pipes with integer datetimes.

- **Fix recursive parameters references.**  
  A single pipe may now reference itself when symlinking parameters.

- **Add `self` syntax for parameters symlinking.**  
  Pipes may now use `self` in parameters rather than requiring the fully qualified name:

  ```python
  import meerschaum as mrsm

  pipe = mrsm.Pipe(
      'sql:memory', 'symlink', 'self',
      instance='sql:memory',
      parameters={
          "custom": {
            "value": 123,
          },
          "sql": "SELECT * FROM foo WHERE val = {{ self.parameters['custom']['value'] }}",
      },
  )

  print(pipe.parameters['sql'])
  # SELECT * FROM foo WHERE val = 123 
  ```

Author: bmeares

CHANGELOG.md

@@ -4,6 +4,35 @@
 
 This is the current release cycle, so stay tuned for future releases!
 
+### v3.1.3
+
+- **Fix syncing pipes with integer datetimes.**  
+  An issue with dtype detection has been fixed, improving performance for pipes with integer datetimes.
+
+- **Fix recursive parameters references.**  
+  A single pipe may now reference itself when symlinking parameters.
+
+- **Add `self` syntax for parameters symlinking.**  
+  Pipes may now use `self` in parameters rather than requiring the fully qualified name:
+
+  ```python
+  import meerschaum as mrsm
+
+  pipe = mrsm.Pipe(
+      'sql:memory', 'symlink', 'self',
+      instance='sql:memory',
+      parameters={
+          "custom": {
+            "value": 123,
+          },
+          "sql": "SELECT * FROM foo WHERE val = {{ self.parameters['custom']['value'] }}",
+      },
+  )
+
+  print(pipe.parameters['sql'])
+  # SELECT * FROM foo WHERE val = 123 
+  ```
+
 ### v3.1.2
 
 - **Improve filtering logic for non-datetime pipes.**  

docs/mkdocs/news/changelog.md

@@ -4,6 +4,35 @@
 
 This is the current release cycle, so stay tuned for future releases!
 
+### v3.1.3
+
+- **Fix syncing pipes with integer datetimes.**  
+  An issue with dtype detection has been fixed, improving performance for pipes with integer datetimes.
+
+- **Fix recursive parameters references.**  
+  A single pipe may now reference itself when symlinking parameters.
+
+- **Add `self` syntax for parameters symlinking.**  
+  Pipes may now use `self` in parameters rather than requiring the fully qualified name:
+
+  ```python
+  import meerschaum as mrsm
+
+  pipe = mrsm.Pipe(
+      'sql:memory', 'symlink', 'self',
+      instance='sql:memory',
+      parameters={
+          "custom": {
+            "value": 123,
+          },
+          "sql": "SELECT * FROM foo WHERE val = {{ self.parameters['custom']['value'] }}",
+      },
+  )
+
+  print(pipe.parameters['sql'])
+  # SELECT * FROM foo WHERE val = 123 
+  ```
+
 ### v3.1.2
 
 - **Improve filtering logic for non-datetime pipes.**  

docs/mkdocs/reference/pipes/parameters.md

@@ -48,6 +48,8 @@ This page catalogs useful keys in the `parameters` dictionary.
             'upsert': "{{ Pipe('demo', 'symlink', instance='sql:memory').upsert }}",
             'custom': "{{ Pipe('demo', 'symlink', instance='sql:memory').parameters['custom'] }}",
             'parent_pipe_id': "{{ Pipe('demo', 'symlink', instance='sql:memory').id }}",
+            'value': 123,
+            'self_value': "{{ self.parameters['value'] }}",
         },
     )
 
@@ -67,6 +69,8 @@ This page catalogs useful keys in the `parameters` dictionary.
     #         ]
     #     },
     #     "parent_pipe_id": 1
+    #     "value": 123,
+    #     "self_value": 123,
     # }  
     ```
 

meerschaum/config/_version.py

@@ -2,4 +2,4 @@
 Specify the Meerschaum release version.
 """
 
-__version__ = "3.1.2"
+__version__ = "3.1.3"

meerschaum/connectors/sql/_sql.py

@@ -624,12 +624,17 @@ def exec(
             warn(str(e), stacklevel=3)
         result = None
         if _commit:
+            if debug:
+                dprint(f"[{self}] Rolling back failed transaction...")
             transaction.rollback()
             connection = self.get_connection(rebuild=True)
     finally:
         if _close:
             connection.close()
 
+    if debug:
+        dprint(f"[{self}] Done executing.")
+
     if with_connection:
         return result, connection
 
@@ -709,8 +714,14 @@ def exec_queries(
                 elif debug:
                     dprint(f"[{self}]\n" + str(msg))
                 result = None
+
+            if debug:
+                dprint(f"[{self}] Finished executing.")
+
             if result is None and break_on_error:
                 if rollback:
+                    if debug:
+                        dprint(f"[{self}] Rolling back...")
                     session.rollback()
                 results.append(result)
                 break

meerschaum/core/Pipe/_attributes.py

@@ -118,7 +118,7 @@ def recursive_replace(obj: Any, path: tuple) -> Any:
         if isinstance(obj, list):
             return [recursive_replace(elem, path + (i,)) for i, elem in enumerate(obj)]
         if isinstance(obj, str):
-            substituted_val = replace_pipes_syntax(obj)
+            substituted_val = replace_pipes_syntax(obj, _pipe=self)
             if substituted_val != obj:
                 self._symlinks[path] = {
                     'original': obj,

meerschaum/utils/dtypes/__init__.py

@@ -222,8 +222,12 @@ def are_dtypes_equal(
         'int', 'int64', 'int32', 'int16', 'int8',
         'uint', 'uint64', 'uint32', 'uint16', 'uint8',
     )
+    int_substrings = ('int',)
     if ldtype.lower() in int_dtypes and rdtype.lower() in int_dtypes:
         return True
+    for substring in int_substrings:
+        if substring in ldtype.lower() and substring in rdtype.lower():
+            return True
 
     float_dtypes = ('float', 'float64', 'float32', 'float16', 'float128', 'double')
     if ldtype.lower() in float_dtypes and rdtype.lower() in float_dtypes:

meerschaum/utils/pipes.py

@@ -18,11 +18,14 @@
 import meerschaum as mrsm
 
 
-def evaluate_pipe_access_chain(access_chain: str, pipe: mrsm.Pipe):
+def evaluate_pipe_access_chain(access_chain: str, pipe: mrsm.Pipe, _pipe=None):
     """
     Safely evaluate the access chain on a Pipe.
     """
     expr = f"pipe{access_chain}"
+    if pipe == _pipe and access_chain.lstrip('.').startswith('parameters'):
+        parameters_access_chain = access_chain.lstrip('.').split('parameters', maxsplit=1)[-1]
+        expr = f"pipe.attributes['parameters']{parameters_access_chain}"
     tree = ast.parse(expr, mode='eval')
 
     def _eval(node, context):
@@ -59,18 +62,22 @@ def _eval(node, context):
 
 
 
-def _evaluate_pipe_access_chain_from_match(pipe_match: re.Match) -> Any:
+def _evaluate_pipe_access_chain_from_match(pipe_match: re.Match, _pipe=None) -> Any:
     """
     Helper function to evaluate a pipe from a regex match object.
     """
     from meerschaum.utils.warnings import warn
     from meerschaum.utils.misc import parse_arguments_str
     from meerschaum.utils.sql import sql_item_name
     try:
-        args_str = pipe_match.group(1)
-        access_chain = pipe_match.group(2)
-        args, kwargs = parse_arguments_str(args_str)
-        pipe = mrsm.Pipe(*args, **kwargs)
+        if 'self' in pipe_match.group(0) and _pipe is not None:
+            pipe = _pipe
+            access_chain = pipe_match.group(1)
+        else:
+            args_str = pipe_match.group(1)
+            access_chain = pipe_match.group(2)
+            args, kwargs = parse_arguments_str(args_str)
+            pipe = mrsm.Pipe(*args, **kwargs)
     except Exception as e:
         warn(f"Failed to parse pipe from template string:\n{e}")
         raise e
@@ -88,18 +95,22 @@ def _evaluate_pipe_access_chain_from_match(pipe_match: re.Match) -> Any:
             else pipe.target
         )
 
-    return evaluate_pipe_access_chain(access_chain, pipe)
+    return evaluate_pipe_access_chain(access_chain, pipe, _pipe=_pipe)
 
 
-def replace_pipes_syntax(text: str) -> Any:
+def replace_pipes_syntax(text: str, _pipe=None) -> Any:
     """
     Parse a string containing the `{{ Pipe() }}` syntax.
     """
     from meerschaum.utils.warnings import warn
     from meerschaum.utils.dtypes import json_serialize_value
     pattern = r'\{\{\s*(?:mrsm\.)?Pipe\((.*?)\)((?:\.[\w]+|\[[^\]]+\])*)\s*\}\}'
+    self_pattern = r'\{\{\s*self((?:\.[\w]+|\[[^\]]+\])*)\s*\}\}'
 
     matches = list(re.finditer(pattern, text))
+    if _pipe is not None:
+        self_matches = list(re.finditer(self_pattern, text))
+        matches.extend(self_matches)
     if not matches:
         return text
 
@@ -115,9 +126,10 @@ def replace_pipes_syntax(text: str) -> Any:
     resolved_values = {}
     for placeholder, match in placeholders.items():
         try:
-            resolved_values[placeholder] = _evaluate_pipe_access_chain_from_match(match)
+            resolved_values[placeholder] = _evaluate_pipe_access_chain_from_match(match, _pipe=_pipe)
         except Exception as e:
-            warn(f"Failed to resolve pipe syntax '{match.group(0)}': {e}")
+            import traceback
+            warn(f"Failed to resolve pipe syntax '{match.group(0)}': {traceback.format_exc()}")
             resolved_values[placeholder] = match.group(0)
 
     if len(matches) == 1: