SQL の Lint ツール SQLFluff にカスタムルールを追加する

database

SQL の Lint ツール sqlfluff にはプラグインでカスタムルールを追加する仕組みがある。

example plugin を参考にプラグインを作る。

BaseRule を継承した Rule_XXXX クラスに、ルールを評価して LintResult を返す _eval() を実装する。

# my_custom_rules/rules.py
from sqlfluff.core.rules import (
    BaseRule,
    LintResult,
    RuleContext,
)
from sqlfluff.core.rules.crawlers import SegmentSeekerCrawler

class Rule_Example_L001(BaseRule):
    '''
    test rule
    '''
    
    groups = ("all",)
    config_keywords = ["aaa"]
    crawl_behaviour = SegmentSeekerCrawler({"select_clause"})
    is_fix_compatible = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        print("---- init ----")
        print(f"parameter: aaa={self.aaa}")

    def _eval(self, context: RuleContext):
        print("---- eval ----")
        for seg in context.segment.segments:
            print(seg.get_type(), seg.raw.lower())
            if seg.get_type() == 'select_clause_element' and seg.raw.lower() == 'b':
                return LintResult(
                    anchor=seg,
                    description=f"Don't select b!",
                )

__init__.py の get_rules() でこのクラスを返す。

# my_custom_rules/__init__.py
from typing import Any, Dict, List, Type

from sqlfluff.core.config import load_config_resource
from sqlfluff.core.plugin import hookimpl
from sqlfluff.core.rules import BaseRule

@hookimpl
def get_rules() -> List[Type[BaseRule]]:
    from my_custom_rules.rules import Rule_Example_L001

    return [Rule_Example_L001]

@hookimpl
def load_default_config() -> Dict[str, Any]:
    """Loads the default configuration for the plugin."""
    return load_config_resource(
        package="my_custom_rules",
        file_name="plugin_default_config.cfg",
    )

@hookimpl
def get_configs_info() -> Dict[str, Dict[str, Any]]:
    """Get rule config validations and descriptions."""
    return {
        "aaa": {"definition": "aaaa"},
    }

あとは pyproject.toml を置いて pip install すると sqlfluff が認識して Lint がはたらく。

$ cat pyproject.toml
[build-system]
requires = ["setuptools>=40.8.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "sqlfluff-my-custom-rules"
version = "1.0.0"
requires-python = ">=3.8"
dependencies = [
    "sqlfluff>=2.2.2"
]

[project.entry-points.sqlfluff]
my_custom_rules = "my_custom_rules"

$ pip install -e .

$ cat test.sql 
SELECT
    a,
    b
FROM table
ORDER BY bar;

$ sqlfluff lint --dialect ansi test.sql
---- init ----
parameter: aaa=bbb
rule Example_L001:   0%|                                                                           | 0/68 [00:00<?, ?it/s]
---- eval ----
keyword select
indent 
newline 

whitespace     
select_clause_element a
comma ,
newline 

whitespace     
select_clause_element b
== [test.sql] FAIL                                                                                                        
L:   3 | P:   5 | Example_L001 | Don't select b!
All Finished 📜 🎉!