diff --git a/airbyte_cdk/sources/declarative/interpolation/filters.py b/airbyte_cdk/sources/declarative/interpolation/filters.py index 4824e3a9f2..185d9dcb31 100644 --- a/airbyte_cdk/sources/declarative/interpolation/filters.py +++ b/airbyte_cdk/sources/declarative/interpolation/filters.py @@ -136,6 +136,13 @@ def regex_search(value: str, regex: str) -> str: return "" +def regex_replace(value: str, regex: str, replacement: str) -> str: + """ + Replace all occurrences of a regular expression pattern in a string. + """ + return re.sub(regex, replacement, value) + + def hmac(value: Any, key: str, hash_type: str = "sha256") -> str: """ Implementation of a custom Jinja2 hmac filter with SHA-256 support. @@ -181,6 +188,7 @@ def hmac(value: Any, key: str, hash_type: str = "sha256") -> str: base64binascii_decode, string, regex_search, + regex_replace, hmac, ] filters = {f.__name__: f for f in _filters_list} diff --git a/unit_tests/sources/declarative/interpolation/test_filters.py b/unit_tests/sources/declarative/interpolation/test_filters.py index a48c6a3010..5ffd2aff35 100644 --- a/unit_tests/sources/declarative/interpolation/test_filters.py +++ b/unit_tests/sources/declarative/interpolation/test_filters.py @@ -108,6 +108,36 @@ def test_regex_search_no_match() -> None: assert val is None +@pytest.mark.parametrize( + "expression, expected", + [ + pytest.param( + "{{ 'hello world' | regex_replace('world', 'there') }}", + "hello there", + id="basic_replacement", + ), + pytest.param( + "{{ 'abc123def456' | regex_replace('[0-9]+', '') }}", + "abcdef", + id="regex_pattern_strip_digits", + ), + pytest.param( + "{{ 'hello world' | regex_replace('xyz', 'replaced') }}", + "hello world", + id="no_match_returns_original", + ), + pytest.param( + "{{ 'aaa bbb aaa' | regex_replace('aaa', 'ccc') }}", + "ccc bbb ccc", + id="multiple_occurrences", + ), + ], +) +def test_regex_replace(expression: str, expected: str) -> None: + val = interpolation.eval(expression, {}) + assert val == expected + + def test_hmac_sha256_default() -> None: message = "test_message" secret_key = "test_secret_key"