How to Safely Mask Table and Column Names Before Sending SQL Queries to AI

import re
import json
from typing import Dict, Tuple

class SqlSchemaMasker:
    """
    Mask table and column names in SQL schemas before AI processing.
    Maintains a reversible mapping to restore original names in AI-generated SQL.
    """

    def __init__(self):
        self.table_map: Dict[str, str] = {}    # real_name → TABLE_XX
        self.column_map: Dict[str, str] = {}   # real_name → col_XX
        self.reverse_map: Dict[str, str] = {}  # masked → real
        self._table_counter = 1
        self._col_counter = 1

    def _mask_table(self, name: str) -> str:
        if name not in self.table_map:
            masked = f"TABLE_{self._table_counter:02d}"
            self.table_map[name] = masked
            self.reverse_map[masked.lower()] = name
            self.reverse_map[masked] = name
            self._table_counter += 1
        return self.table_map[name]

    def _mask_column(self, name: str) -> str:
        if name not in self.column_map:
            masked = f"col_{self._col_counter:02d}"
            self.column_map[name] = masked
            self.reverse_map[masked.lower()] = name
            self.reverse_map[masked] = name
            self._col_counter += 1
        return self.column_map[name]

    def mask_schema(self, sql: str) -> str:
        """Mask a full schema (one or more CREATE TABLE statements)."""
        # Mask CREATE TABLE statements
        sql = re.sub(
            r'CREATEs+TABLEs+(?:IFs+NOTs+EXISTSs+)?[`"]?(w+)[`"]?',
            lambda m: f"CREATE TABLE {self._mask_table(m.group(1))}",
            sql, flags=re.IGNORECASE
        )
        # Mask column names in column definitions
        # Pattern: leading whitespace + column_name + data_type
        sql = re.sub(
            r'^(s+)[`"]?(w+)[`"]?s+(BIGINT|INT|INTEGER|SMALLINT|TINYINT'
            r'|VARCHAR|CHAR|TEXT|LONGTEXT|MEDIUMTEXT'
            r'|DECIMAL|NUMERIC|FLOAT|DOUBLE|REAL'
            r'|BOOLEAN|BOOL|DATE|DATETIME|TIMESTAMP|TIME'
            r'|JSON|BLOB|BINARY)',
            lambda m: f"{m.group(1)}{self._mask_column(m.group(2))} {m.group(3)}",
            sql, flags=re.IGNORECASE | re.MULTILINE
        )
        # Mask FOREIGN KEY references
        sql = re.sub(
            r'FOREIGNs+KEYs*([`"]?(w+)[`"]?)',
            lambda m: f"FOREIGN KEY ({self.column_map.get(m.group(1), m.group(1))})",
            sql, flags=re.IGNORECASE
        )
        # Mask REFERENCES table(column)
        sql = re.sub(
            r'REFERENCESs+[`"]?(w+)[`"]?s*([`"]?(w+)[`"]?)',
            lambda m: (
                f"REFERENCES {self.table_map.get(m.group(1), m.group(1))}"
                f"({self.column_map.get(m.group(2), m.group(2))})"
            ),
            sql, flags=re.IGNORECASE
        )
        return sql

    def unmask_sql(self, sql: str) -> str:
        """Replace masked names back to originals in AI-generated SQL."""
        result = sql
        # Sort by length descending to avoid partial replacements (col_10 before col_1)
        for masked, original in sorted(
            self.reverse_map.items(), key=lambda x: -len(x[0])
        ):
            result = re.sub(
                r'' + re.escape(masked) + r'',
                original,
                result,
                flags=re.IGNORECASE
            )
        return result

    def export_mapping(self) -> str:
        """Export the masking map as JSON for reference or storage."""
        return json.dumps({
            "tables": self.table_map,
            "columns": self.column_map
        }, indent=2)

    @classmethod
    def from_mapping(cls, mapping_json: str) -> 'SqlSchemaMasker':
        """Restore a masker from a previously exported mapping."""
        masker = cls()
        mapping = json.loads(mapping_json)
        masker.table_map = mapping["tables"]
        masker.column_map = mapping["columns"]
        # Build reverse map
        for real, masked in mapping["tables"].items():
            masker.reverse_map[masked.lower()] = real
            masker.reverse_map[masked] = real
        for real, masked in mapping["columns"].items():
            masker.reverse_map[masked.lower()] = real
            masker.reverse_map[masked] = real
        return masker


# ─── Example Usage ───────────────────────────────────────────────────────────
masker = SqlSchemaMasker()

original_schema = """
CREATE TABLE customer_churn_risk (
  customer_id BIGINT PRIMARY KEY,
  email VARCHAR(255),
  churn_score DECIMAL(5,2),
  revenue_ltv DECIMAL(10,2),
  health_score INT,
  is_at_risk BOOLEAN,
  last_login_date DATE
);

CREATE TABLE competitor_tracking (
  entry_id BIGINT PRIMARY KEY,
  customer_id BIGINT,
  competitor_name VARCHAR(100),
  switch_date DATE,
  FOREIGN KEY (customer_id) REFERENCES customer_churn_risk(customer_id)
);
"""

masked = masker.mask_schema(original_schema)
print("=== SEND TO AI ===")
print(masked)

# Save mapping for later use
mapping = masker.export_mapping()
print("
=== SAVE THIS MAPPING ===")
print(mapping)

# After AI generates a query, restore original names:
ai_query = """
SELECT col_01, col_03, col_07
FROM TABLE_01
WHERE col_07 = true
  AND col_04 > 1000
ORDER BY col_03 DESC;
"""

restored = masker.unmask_sql(ai_query)
print("
=== RESTORED QUERY ===")
print(restored)
# SELECT customer_id, churn_score, is_at_risk
# FROM customer_churn_risk
# WHERE is_at_risk = true
#   AND revenue_ltv > 1000
# ORDER BY churn_score DESC;