feat(jinja): add b64decode/b64encode builtin filters

Core SQLMesh shipped no builtin Jinja filters, so base64-encoded secrets
(e.g. a BigQuery service-account key stored in an env var) could not be
decoded directly in config YAML. Add b64decode/b64encode and register them
on the shared environment factory so they are available both in config YAML
and in models, mirroring ansible's filters.

Closes #5754

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

sqlmesh/utils/jinja.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import importlib
 import json
 import re
@@ -28,11 +29,30 @@
 SQLMESH_JINJA_PACKAGE = "sqlmesh.utils.jinja"
 
 
+def b64decode(value: t.Union[str, bytes]) -> str:
+    decoded = value.encode("utf-8") if isinstance(value, str) else value
+    return base64.b64decode(decoded).decode("utf-8")
+
+
+def b64encode(value: t.Union[str, bytes]) -> str:
+    encoded = value.encode("utf-8") if isinstance(value, str) else value
+    return base64.b64encode(encoded).decode("utf-8")
+
+
+def create_builtin_filters() -> t.Dict[str, t.Callable]:
+    return {
+        "b64decode": b64decode,
+        "b64encode": b64encode,
+    }
+
+
 def environment(**kwargs: t.Any) -> Environment:
     extensions = kwargs.pop("extensions", [])
     extensions.append("jinja2.ext.do")
     extensions.append("jinja2.ext.loopcontrols")
-    return Environment(extensions=extensions, **kwargs)
+    env = Environment(extensions=extensions, **kwargs)
+    env.filters.update(create_builtin_filters())
+    return env
 
 
 ENVIRONMENT = environment()

tests/utils/test_jinja.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from sqlmesh.utils import AttributeDict
+from base64 import b64encode
+
+from sqlmesh.utils import AttributeDict, yaml
 from sqlmesh.utils.jinja import (
     ENVIRONMENT,
     JinjaMacroRegistry,
@@ -329,3 +331,16 @@ def test_macro_registry_to_expressions_sorted():
         == "refs = {'orders': {'database': 'jaffle_shop', 'nested_list': ['a', 'b', 'c'], 'schema': 'main'}, 'payments': {'database': 'jaffle_shop', 'nested': {'baz': 'bing', 'foo': 'bar'}, 'schema': 'main'}}\n"
         "sources = {}"
     )
+
+
+def test_builtin_base64_filters():
+    encoded = b64encode(b"secret").decode("utf-8")
+
+    env = JinjaMacroRegistry().build_environment()
+    assert env.from_string("{{ value | b64decode }}").render(value=encoded) == "secret"
+    assert env.from_string("{{ 'secret' | b64encode }}").render() == encoded
+    assert env.from_string("{{ 'secret' | b64encode | b64decode }}").render() == "secret"
+
+    # The same filters are available when rendering Jinja in config YAML files.
+    config = yaml.load(f'env_vars:\n  TOKEN: "{{{{ "{encoded}" | b64decode }}}}"')
+    assert config == {"env_vars": {"TOKEN": "secret"}}