feat(python cdk): Allow regex_search in jinja interpolations (#40696)

Co-authored-by: Octavia Squidington III <[email protected]>

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2600,3 +2600,10 @@ interpolation:
       examples:
         - '{{ 1 | string }} -> "1"'
         - '{{ ["hello", "world" | string }} -> "["hello", "world"]"'
+    - title: Regex Search
+      description: Match the input string against a regular expression and return the first match.
+      arguments:
+        regex: The regular expression to search for. It must include a capture group.
+      return_type: str
+      examples:
+        - '{{ "goodbye, cruel world" | regex_search("goodbye,\s(.*)$") }} -> "cruel world"'

airbyte-cdk/python/airbyte_cdk/sources/declarative/interpolation/filters.py

@@ -4,6 +4,7 @@
 import base64
 import hashlib
 import json
+import re
 from typing import Any, Optional
 
 
@@ -105,5 +106,15 @@ def string(value: Any) -> str:
     return ret
 
 
-_filters_list = [hash, base64encode, base64decode, string]
+def regex_search(value: str, regex: str) -> str:
+    """
+    Match a regular expression against a string and return the first match group if it exists.
+    """
+    match = re.search(regex, value)
+    if match and len(match.groups()) > 0:
+        return match.group(1)
+    return ""
+
+
+_filters_list = [hash, base64encode, base64decode, string, regex_search]
 filters = {f.__name__: f for f in _filters_list}

airbyte-cdk/python/airbyte_cdk/sources/declarative/interpolation/jinja.py

@@ -97,7 +97,7 @@ def eval(
                     return self._literal_eval(result, valid_types)
             else:
                 # If input is not a string, return it as is
-                raise Exception(f"Expected a string. got {input_str}")
+                raise Exception(f"Expected a string, got {input_str}")
         except UndefinedError:
             pass
         # If result is empty or resulted in an undefined error, evaluate and return the default string

airbyte-cdk/python/unit_tests/sources/declarative/interpolation/test_filters.py

@@ -10,7 +10,7 @@
 interpolation = JinjaInterpolation()
 
 
-def test_hash_md5_no_salt():
+def test_hash_md5_no_salt() -> None:
     input_string = "abcd"
     s = "{{ '%s' | hash('md5') }}" % input_string
     filter_hash = interpolation.eval(s, config={})
@@ -23,7 +23,7 @@ def test_hash_md5_no_salt():
     assert filter_hash == hashlib_computed_hash
 
 
-def test_hash_md5_on_numeric_value():
+def test_hash_md5_on_numeric_value() -> None:
     input_value = 123.456
     s = "{{ %f | hash('md5') }}" % input_value
     filter_hash = interpolation.eval(s, config={})
@@ -36,7 +36,7 @@ def test_hash_md5_on_numeric_value():
     assert filter_hash == hashlib_computed_hash
 
 
-def test_hash_md5_with_salt():
+def test_hash_md5_with_salt() -> None:
     input_string = "test_input_string"
     input_salt = "test_input_salt"
 
@@ -55,7 +55,7 @@ def test_hash_md5_with_salt():
     "input_string",
     ["test_input_client_id", "some_client_secret_1", "12345", "775.78"],
 )
-def test_base64encode(input_string: str):
+def test_base64encode(input_string: str) -> None:
     s = "{{ '%s' | base64encode }}" % input_string
     filter_base64encode = interpolation.eval(s, config={})
 
@@ -73,8 +73,32 @@ def test_base64encode(input_string: str):
         ("cGFzc3dvcmQ=", "password"),
     ],
 )
-def test_base64decode(input_string: str, expected_string: str):
+def test_base64decode(input_string: str, expected_string: str) -> None:
     s = "{{ '%s' | base64decode }}" % input_string
     filter_base64decode = interpolation.eval(s, config={})
 
     assert filter_base64decode == expected_string
+
+
+def test_regex_search_valid() -> None:
+    expression_with_regex = "{{ '<https://this-is-test-link.com/?page=2>; rel=\"next\"' | regex_search('<(.*)>; rel=.*') }}"
+
+    val = interpolation.eval(expression_with_regex, {})
+    assert val == "https://this-is-test-link.com/?page=2"
+
+
+def test_regex_search_no_match_group() -> None:
+    # If no group is set in the regular expression, the result will be an empty string
+    expression_with_regex = "{{ '<https://this-is-test-link.com/?page=2>; rel=\"next\"' | regex_search('<.*>; rel=.*') }}"
+
+    val = interpolation.eval(expression_with_regex, {})
+    assert val is None
+
+
+def test_regex_search_no_match() -> None:
+    # If no group is set in the regular expression, the result will be an empty string
+    expression_with_regex = "{{ '<https://this-is-test-link.com/?page=2>; rel=\"next\"' | regex_search('WATWATWAT') }}"
+
+    val = interpolation.eval(expression_with_regex, {})
+
+    assert val is None