Semantic Rules Guide¶

from fastssv import validate_sql_structured

# Run all rules (recommended API)
violations = validate_sql_structured(sql_query)
for v in violations:
    print(f"{v.severity.value}: [{v.rule_id}] {v.message}")
    print(f"  Fix: {v.suggested_fix}")

# Run only concept standardization rules
violations = validate_sql_structured(sql_query, categories=["concept_standardization"])

# Run specific rule
violations = validate_sql_structured(
    sql_query,
    rule_ids=["concept_standardization.standard_concept_enforcement"]
)

# Run grouped validation results
from fastssv import validate_sql
results = validate_sql(sql_query, categories=["concept_standardization"])
print(results["category_errors"]["concept_standardization"])

STANDARD_CONCEPT_FIELDS = {
    ("condition_occurrence", "condition_concept_id"),
    ("drug_exposure", "drug_concept_id"),
    ("drug_exposure", "route_concept_id"),
    ("procedure_occurrence", "procedure_concept_id"),
    ("measurement", "measurement_concept_id"),
    ("observation", "observation_concept_id"),
    ("visit_occurrence", "visit_concept_id"),
    ("device_exposure", "device_concept_id"),
    # ... 50+ fields total
}

# src/fastssv/rules/concept_standardization/concept_domain_validation.py

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

EXPECTED_DOMAINS = {
    ("drug_exposure", "drug_concept_id"): "Drug",
    ("condition_occurrence", "condition_concept_id"): "Condition",
    ("measurement", "measurement_concept_id"): "Measurement",
    ("procedure_occurrence", "procedure_concept_id"): "Procedure",
    ("observation", "observation_concept_id"): "Observation",
    ("device_exposure", "device_concept_id"): "Device",
}

@register
class DomainValidationRule(Rule):
    """Ensures concepts belong to the correct domain for their table.

    OMOP CDM concept.domain_id must match the table's expected domain.
    For example, drug_exposure.drug_concept_id should only reference
    concepts with domain_id = 'Drug'.
    """

    rule_id = "concept_standardization.concept_domain_validation"
    name = "Concept Domain Validation"
    description = "Validates concept domains match table expectations"
    severity = Severity.WARNING
    suggested_fix = "Filter by concept.domain_id or use proper vocabulary mapping"

    def validate(self, sql: str, dialect: str = "postgres") -> list[RuleViolation]:
        violations = []
        trees, parse_error = parse_sql(sql, dialect)

        if parse_error:
            return violations

        tree = trees[0]
        aliases = extract_aliases(tree)

        # Extract concept field references
        concept_refs = self._extract_concept_references(tree, aliases)

        for table, column, location in concept_refs:
            expected_domain = EXPECTED_DOMAINS.get((table, column))

            if expected_domain:
                # Check if query filters by domain_id
                has_domain_filter = self._has_domain_filter(tree, expected_domain, aliases)

                if not has_domain_filter:
                    violations.append(
                        RuleViolation(
                            rule_id=self.rule_id,
                            severity=self.severity,
                            message=f"Query uses {table}.{column} but does not filter "
                                   f"by concept.domain_id = '{expected_domain}'. This may "
                                   f"include concepts from incorrect domains.",
                            suggested_fix=f"Add JOIN to concept table with WHERE clause: "
                                        f"concept.domain_id = '{expected_domain}'",
                            location=location,
                            details={
                                "table": table,
                                "column": column,
                                "expected_domain": expected_domain
                            }
                        )
                    )

        return violations

    def _extract_concept_references(self, tree, aliases):
        """Extract references to concept fields."""
        # Implementation: scan WHERE/JOIN clauses for concept_id columns
        return []

    def _has_domain_filter(self, tree, domain, aliases):
        """Check if query filters by concept.domain_id."""
        # Implementation: check for domain_id = 'domain' in WHERE/JOIN
        return False

from .concept_domain_validation import ConceptDomainValidationRule

# tests/test_domain_validation.py

import pytest
from fastssv.rules.concept_standardization.concept_domain_validation import ConceptDomainValidationRule

class TestDomainValidation:
    @pytest.fixture
    def rule(self):
        return ConceptDomainValidationRule()

    def test_valid_domain_filter(self, rule):
        sql = """
        SELECT * FROM drug_exposure de
        JOIN concept c ON de.drug_concept_id = c.concept_id
        WHERE c.domain_id = 'Drug'
        """
        violations = rule.validate(sql)
        assert len(violations) == 0

    def test_missing_domain_filter(self, rule):
        sql = """
        SELECT * FROM drug_exposure de
        JOIN concept c ON de.drug_concept_id = c.concept_id
        WHERE c.concept_name LIKE '%aspirin%'
        """
        violations = rule.validate(sql)
        assert len(violations) > 0
        assert violations[0].rule_id == "concept_standardization.concept_domain_validation"

from fastssv import get_all_rules
for rule_cls in get_all_rules():
    rule = rule_cls()
    print(rule.rule_id, rule.severity.name)

RuleViolation(
    rule_id=self.rule_id,
    severity=Severity.ERROR,
    message="What's wrong and why it matters. Query uses X without Y, "
           "which may cause Z problem.",
    suggested_fix="Specific actionable fix. Add WHERE clause: X = Y OR "
                  "use JOIN to table Z with condition A = B",
    location="table.column or specific SQL fragment",
    details={
        "table": "condition_occurrence",
        "field": "condition_concept_id",
        "additional_context": "any structured data"
    }
)

from fastssv.core.helpers import (
    parse_sql,              # Parse SQL to AST
    extract_aliases,        # Map aliases to table names
    resolve_table_col,      # Resolve column to (table, column)
    normalize_name,         # Case-insensitive name comparison
    has_table_reference,    # Check if query uses a table
    extract_join_conditions # Extract JOIN conditions
)

def validate(self, sql: str, dialect: str = "postgres") -> list[RuleViolation]:
    violations = []
    trees, parse_error = parse_sql(sql, dialect)

    if parse_error:
        # Don't fail validation on parse errors
        return violations

    # Validation logic
    return violations

violations = validate_sql_structured(sql, categories=["concept_standardization"])
for v in violations:
    print(f"\nRule: {v.rule_id}")
    print(f"Severity: {v.severity.value}")
    print(f"Issue: {v.message}")
    print(f"Fix: {v.suggested_fix}")
    if v.location:
        print(f"Location: {v.location}")
    if v.details:
        print(f"Details: {v.details}")

from fastssv.rules.concept_standardization.standard_concept_enforcement import StandardConceptEnforcementRule

rule = StandardConceptEnforcementRule()
violations = rule.validate(sql, dialect="postgres")

for v in violations:
    print(v.to_dict())

from fastssv.core.helpers import parse_sql

trees, error = parse_sql(sql, dialect="postgres")
if not error:
    print(trees[0].sql())  # Pretty-print parsed SQL
    print(trees[0])        # Print AST structure