Replace regular expression in VIEW converter with sqlglot tokenizer

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [0.60.0] - 2025-11-11
+
+- Replaced regular expression in `VIEW` converter with sqlglot tokenizer.
+- Introduced new category for optional dependencies: "convert". Example: `pip install snowddl[convert]`. It is only required if you plan to use converters.
+
 ## [0.59.1] - 2025-10-25
 
 - Fixed issue with optional `mfa_enrollment` parameter for `AUTHENTICATION_POLICY`.

pyproject.toml

@@ -36,6 +36,9 @@ file = "README.md"
 content-type = "text/markdown"
 
 [project.optional-dependencies]
+convert = [
+    "sqlglot"
+]
 dev = [
     "black",
     "pytest",

snowddl/converter/view.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from re import compile, DOTALL
+from sqlglot import Tokenizer, TokenType
 
 from snowddl.blueprint import ObjectType
 from snowddl.converter.abc_converter import ConvertResult
@@ -8,9 +8,6 @@
 from snowddl.parser.view import view_json_schema
 
 
-view_text_re = compile(r"^.*?\sas(\n|\s)+(.*)$", DOTALL)
-
-
 class ViewConverter(AbstractSchemaObjectConverter):
     def get_object_type(self) -> ObjectType:
         return ObjectType.VIEW
@@ -77,13 +74,8 @@ def _get_columns(self, row):
         return cols
 
     def _get_text_or_include(self, object_path: Path, row: dict):
-        # TODO: replace with better implementation when available
-        cur = self.engine.execute_meta(
-            "SELECT GET_DDL('VIEW', {ident}) AS view_ddl", {"ident": f"{row['database']}.{row['schema']}.{row['name']}"}
-        )
-
-        view_ddl = cur.fetchone()["VIEW_DDL"]
-        view_text = view_text_re.sub(r"\2", view_ddl).rstrip(";").strip(" \n\r\t")
+        view_text = self._extract_view_text_after_as(row)
+        view_text = view_text.rstrip(";").strip(" \n\r\t")
 
         # Remove trailing spaces from each line to prevent output formatting issues
         # https://github.com/yaml/pyyaml/issues/411
@@ -96,3 +88,12 @@ def _get_text_or_include(self, object_path: Path, row: dict):
             return YamlIncludeStr(f"sql/{file_name}")
 
         return YamlLiteralStr(view_text)
+
+    def _extract_view_text_after_as(self, row: dict):
+        tokenizer = Tokenizer()
+
+        for token in tokenizer.tokenize(row["text"]):
+            if token.token_type == TokenType.ALIAS:
+                return row["text"][token.end+2:]
+
+        raise ValueError(f"Could not extract view text after 'AS' keyword using sqlglot tokenizer")

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.59.1"
+__version__ = "0.60.0"