FastSSV Plugin Architecture¶

from abc import ABC, abstractmethod
from typing import List

class Rule(ABC):
    """Abstract base class for validation rules."""

    rule_id: str
    name: str
    description: str
    severity: Severity
    suggested_fix: str

    @abstractmethod
    def validate(self, sql: str, dialect: str = "postgres") -> List[RuleViolation]:
        """Validate SQL query and return violations."""
        pass

from dataclasses import dataclass, field
from enum import Enum

class Severity(Enum):
    """Severity level for rule violations."""
    ERROR = "error"      # Blocks validation (gates the CLI exit code)
    WARNING = "warning"  # Best-practice issue; escalated under --strict

@dataclass
class RuleViolation:
    rule_id: str                    # e.g., "concept_standardization.standard_concept_enforcement"
    severity: Severity              # ERROR or WARNING
    message: str                    # Human-readable issue description
    suggested_fix: str              # Prose fix recommendation
    location: Optional[str] = None  # Optional location info (table.column or SQL fragment)
    details: dict = field(default_factory=dict)  # In-process metadata; not serialised
    suggested_fix_patch: Optional[dict] = None   # Mechanical patch (REPLACE/ADD/REMOVE) when applicable

def to_dict(self) -> dict:
    """Serialise for the CLI report and the HTTP API.

    The single ``fix`` field unifies prose and mechanical patches:
    - For FREEFORM patches (the default for ~60% of rules), ``fix`` is the
      prose ``suggested_fix`` string.
    - For REPLACE/ADD/REMOVE patches (~40%), ``fix`` is a patch dict
      ``{"action": ..., "span": [s, e] | "at": pos, "text": ...}``.
    Consumers switch on ``isinstance(fix, str)`` vs ``dict``.

    ``details`` stays on the dataclass for tests and rule-side
    diagnostics but is intentionally excluded from the wire format —
    the keys vary too much across rules to be programmatically useful.
    """
    result = {
        "rule_id": self.rule_id,
        "severity": self.severity.value,   # "error" | "warning"
        "issue": self.message,
    }
    fix = self._serialised_fix()           # prose string OR patch dict
    if fix is not None:
        result["fix"] = fix
    if self.location:
        result["location"] = self.location
    return result

from typing import Type

# Global registry
_registry: dict[str, Type[Rule]] = {}

def register(cls: Type[Rule]) -> Type[Rule]:
    """Decorator to register a rule class by its rule_id attribute."""
    _registry[cls.rule_id] = cls
    return cls

def get_rule(rule_id: str) -> Type[Rule]:
    """Get a rule class by ID."""
    return _registry[rule_id]

def get_rules_by_category(category: str) -> list[Type[Rule]]:
    """Get all rules in a category."""
    prefix = category + "."
    return [r for r in _registry.values() if r.rule_id.startswith(prefix)]

def get_all_rules() -> list[Type[Rule]]:
    """Get all registered rules."""
    return list(_registry.values())

# src/fastssv/rules/joins/my_new_rule.py

from fastssv.core.base import Rule, RuleViolation, Severity
from fastssv.core.registry import register
from fastssv.core.helpers import parse_sql

@register
class MyNewRule(Rule):
    """
    Detailed documentation of the rule.

    Explains:
    - What OMOP CDM constraint this validates
    - Why this constraint matters
    - Examples of violations
    - How to fix violations
    """

    rule_id = "joins.my_new_rule"
    name = "My New Rule"
    description = "Brief description of what this rule checks"
    severity = Severity.ERROR
    suggested_fix = "Specific recommendation for fixing this violation"

    def validate(self, sql: str, dialect: str = "postgres") -> list[RuleViolation]:
        """
        Validate SQL query against this rule.

        Args:
            sql: SQL query string to validate
            dialect: SQL dialect (postgres, mysql, duckdb, etc.)

        Returns:
            List of RuleViolation objects. Empty list means no violations.
        """
        violations = []

        # Parse SQL into AST
        trees, parse_error = parse_sql(sql, dialect)
        if parse_error:
            # Skip validation if SQL doesn't parse
            return violations

        # Validation logic here
        if self._detect_violation(trees[0]):
            violations.append(
                RuleViolation(
                    rule_id=self.rule_id,
                    severity=Severity.ERROR,  # or Severity.WARNING
                    message="Clear explanation of what went wrong",
                    suggested_fix="Specific recommendation for fixing this violation",
                    location="optional: table.column or line number",  # Optional
                    details={"key": "value"}  # Optional structured metadata
                )
            )

        return violations

    def _detect_violation(self, tree) -> bool:
        """Helper method for violation detection logic."""
        # Implementation details
        return False

# src/fastssv/rules/joins/__init__.py

from .my_new_rule import MyNewRule

# tests/test_rules.py — append a new class

import pytest
from fastssv.rules.joins.my_new_rule import MyNewRule

class TestMyNewRule:

    @pytest.fixture
    def rule(self):
        return MyNewRule()

    def test_valid_query_passes(self, rule):
        sql = "SELECT * FROM person"
        violations = rule.validate(sql)
        assert len(violations) == 0

    def test_invalid_query_fails(self, rule):
        sql = "SELECT * FROM invalid_table"
        violations = rule.validate(sql)
        assert len(violations) > 0
        assert violations[0].rule_id == "joins.my_new_rule"

from fastssv import validate_sql_structured

# Run all rules
violations = validate_sql_structured(sql)

# Run specific category
violations = validate_sql_structured(sql, categories=["joins"])

# Run specific rules
violations = validate_sql_structured(
    sql,
    rule_ids=["joins.my_new_rule", "anti_patterns.concept_name_lookup"]
)

# Process violations
for v in violations:
    print(f"{v.severity}: [{v.rule_id}] {v.message}")

RuleViolation(
    rule_id="concept_standardization.standard_concept_enforcement",
    severity=Severity.ERROR,
    message="Query uses STANDARD concept field 'drug_concept_id' without enforcing "
            "standard_concept = 'S'. This may include non-standard concepts in results.",
    suggested_fix="Add WHERE clause: concept.standard_concept = 'S' OR use concept_relationship "
                  "with relationship_id = 'Maps to'"
)

from fastssv.core.helpers import parse_sql

# Parse SQL into sqlglot AST
trees, parse_error = parse_sql(sql, dialect="postgres")

if parse_error:
    # Handle parse error
    return []

# Work with AST
tree = trees[0]

from sqlglot import exp

tables = [
    table.name
    for table in tree.find_all(exp.Table)
]

joins = list(tree.find_all(exp.Join))
for join in joins:
    # Process join
    pass

where = tree.find(exp.Where)
if where:
    # Process WHERE clause
    pass