feat: respect sqlfluff configuration file

sqllineage/core/parser/sqlfluff/analyzer.py

@@ -1,7 +1,13 @@
 import warnings
 from typing import Dict, List
 
-from sqlfluff.core import Linter, SQLLexError, SQLParseError, dialect_readout
+from sqlfluff.core import (
+    FluffConfig,
+    Linter,
+    SQLLexError,
+    SQLParseError,
+    dialect_readout,
+)
 from sqlfluff.core.parser import BaseSegment
 
 from sqllineage.core.analyzer import LineageAnalyzer
@@ -73,7 +79,9 @@ def analyze(
                     )
 
     def _list_specific_statement_segment(self, sql: str):
-        parsed = Linter(dialect=self._dialect).parse_string(sql)
+        parsed = Linter(
+            config=FluffConfig.from_root(overrides={"dialect": self._dialect})
+        ).parse_string(sql)
         violations = [
             str(e)
             for e in parsed.violations

tests/core/test_config.py

@@ -11,8 +11,8 @@
 from sqllineage.runner import LineageRunner
 
 
-@patch(
-    "os.environ",
+@patch.dict(
+    os.environ,
     {
         "SQLLINEAGE_DIRECTORY": os.path.join(os.path.dirname(__file__), "data"),
         "SQLLINEAGE_DEFAULT_SCHEMA": "<default>",

tests/core/test_exception.py

@@ -1,3 +1,4 @@
+import os
 from unittest.mock import patch
 
 import pytest
@@ -56,7 +57,7 @@ def test_syntax_warning_no_semicolon_in_tsql():
         )._eval()
 
 
-@patch("os.environ", {"SQLLINEAGE_TSQL_NO_SEMICOLON": "TRUE"})
+@patch.dict(os.environ, {"SQLLINEAGE_TSQL_NO_SEMICOLON": "TRUE"})
 def test_user_warning_enable_tsql_no_semicolon_with_other_dialect():
     with pytest.warns(UserWarning):
         LineageRunner(

tests/core/test_runner.py

@@ -1,6 +1,10 @@
+import os
+import tempfile
+
 from sqllineage.core.models import SubQuery
 from sqllineage.runner import LineageRunner
 from sqllineage.utils.constant import LineageLevel
+from ..helpers import assert_table_lineage_equal
 
 
 def test_runner_dummy():
@@ -31,3 +35,19 @@ def test_get_column_lineage_exclude_subquery_inpath():
     for col_tuple in parse.get_column_lineage(exclude_subquery_columns=True):
         for col in col_tuple:
             assert not isinstance(col.parent, SubQuery)
+
+
+def test_respect_sqlfluff_configuration_file():
+    sqlfluff_config = """[sqlfluff:templater:jinja:context]
+num_things=456
+tbl_name=my_table"""
+    cwd = os.getcwd()
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        try:
+            os.chdir(tmpdirname)
+            with open(".sqlfluff", "w") as f:
+                f.write(sqlfluff_config)
+            sql = "SELECT {{ num_things }} FROM {{ tbl_name }} WHERE id > 10 LIMIT 5"
+            assert_table_lineage_equal(sql, {"my_table"}, test_sqlparse=False)
+        finally:
+            os.chdir(cwd)

tests/sql/column/test_column_select_lateral_alias_ref.py

@@ -1,3 +1,4 @@
+import os
 from unittest.mock import patch
 
 import pytest
@@ -14,7 +15,7 @@
 
 
 @pytest.mark.parametrize("provider", providers)
-@patch("os.environ", {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
+@patch.dict(os.environ, {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
 def test_column_top_level_enable_lateral_ref(
     provider: MetaDataProvider,
 ):
@@ -106,7 +107,7 @@ def test_column_top_level_enable_lateral_ref(
 
 
 @pytest.mark.parametrize("provider", providers)
-@patch("os.environ", {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
+@patch.dict(os.environ, {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
 def test_column_top_level_enable_lateral_ref_with_metadata_from_table(
     provider: MetaDataProvider,
 ):
@@ -148,7 +149,7 @@ def test_column_top_level_enable_lateral_ref_with_metadata_from_table(
 
 
 @pytest.mark.parametrize("provider", providers)
-@patch("os.environ", {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
+@patch.dict(os.environ, {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
 def test_column_top_level_enable_lateral_ref_with_metadata_from_subquery(
     provider: MetaDataProvider,
 ):
@@ -195,7 +196,7 @@ def test_column_top_level_enable_lateral_ref_with_metadata_from_subquery(
 
 
 @pytest.mark.parametrize("provider", providers)
-@patch("os.environ", {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
+@patch.dict(os.environ, {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
 def test_column_top_level_enable_lateral_ref_with_metadata_from_nested_subquery(
     provider: MetaDataProvider,
 ):
@@ -247,7 +248,7 @@ def test_column_top_level_enable_lateral_ref_with_metadata_from_nested_subquery(
 
 
 @pytest.mark.parametrize("provider", providers)
-@patch("os.environ", {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
+@patch.dict(os.environ, {"SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE": "1"})
 def test_column_enable_lateral_ref_within_subquery(
     provider: MetaDataProvider,
 ):

tests/sql/table/multiple_statements/test_tsql_no_semicolon.py

@@ -1,11 +1,12 @@
+import os
 from unittest.mock import patch
 
 import pytest
 
 from ....helpers import assert_table_lineage_equal
 
 
-@patch("os.environ", {"SQLLINEAGE_TSQL_NO_SEMICOLON": "TRUE"})
+@patch.dict(os.environ, {"SQLLINEAGE_TSQL_NO_SEMICOLON": "TRUE"})
 @pytest.mark.parametrize("dialect", ["tsql"])
 def test_tsql_multi_statement_no_semicolon(dialect: str):
     """