sqldrift 0.1.5

Detect schema drift before your SQL queries blow up.

sqldrift

Detect schema drift before your SQL queries blow up.

sqldrift validates SQL queries against a live database schema to catch missing tables before execution. Built on sqlglot for robust, multi-dialect SQL parsing.

Why?

When AI agents or automated pipelines generate SQL, they may reference tables that have been renamed, dropped, or migrated. Running those queries blindly causes production failures. sqldrift catches the problem before it happens.

Features

• Schema drift detection — identifies missing tables before query execution
• CTE-aware — correctly ignores Common Table Expressions
• Case-insensitive matching (configurable)
• Schema-qualified name support — handles schema.table references
• Optimized for scale — class-based validator with O(1) lookups for 4,000+ tables
• LRU caching — up to 282x speedup for repeated queries
• Multi-dialect — PostgreSQL, MySQL, BigQuery, and more via sqlglot

Installation

pip install .

For development:

pip install -e ".[dev]"

Quick Start

Simple one-off validation

from sqldrift import validate_query

live_tables = ["users", "orders", "products"]

success, message = validate_query(
    "SELECT * FROM users JOIN orders ON users.id = orders.user_id",
    live_tables,
)
# (True, "Query is safe to execute.")

success, message = validate_query(
    "SELECT * FROM deleted_table",
    live_tables,
)
# (False, "Schema Drift Detected: The following tables were not found: ['deleted_table']")

Reusable validator (recommended for large schemas)

from sqldrift import SchemaValidator

validator = SchemaValidator(live_tables)

# Validate many queries — the table set is built once
success, message = validator.validate("SELECT * FROM users")

Column-Level Drift Detection

While the standard validator checks for missing tables, ColumnValidator checks for missing columns. This requires providing a schema definition with column names.

from sqldrift import ColumnValidator

# Define your schema (table -> columns)
schema = {
    "users": {
        "columns": ["id", "name", "email", "created_at"],
        "types": ["INTEGER", "VARCHAR", "VARCHAR", "TIMESTAMP"], # optional
    },
    "orders": {
        "columns": ["id", "user_id", "total"],
    },
}

validator = ColumnValidator(schema)

# Valid query
success, msg = validator.validate("SELECT name FROM users")
# (True, "All columns exist.")

# Invalid query (column 'tier' missing)
success, msg = validator.validate("SELECT tier FROM users")
# (False, "Column Drift Detected: The following columns were not found: ['tier']")

Features

• Qualified names: Handles table.column and alias references (u.name) correctly.
• Suggestions: Offers specific column names if a mismatch is close (e.g. user_id vs userid).
• Caching: Use CachedColumnValidator for high-performance repeated validation.

Cached validator (best for repeated queries)

from sqldrift import CachedSchemaValidator

cached = CachedSchemaValidator(live_tables, cache_size=256)
success, message = cached.validate(query)  # subsequent identical calls are cached

Where Do Table Lists Come From?

sqldrift does not connect to a database — you provide the list of tables that currently exist in your schema. This keeps the library database-agnostic and flexible.

Common ways to source your table list:

# PostgreSQL via psycopg2
cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'")
live_tables = [row[0] for row in cur.fetchall()]

# MySQL via mysql-connector
cur.execute("SHOW TABLES")
live_tables = [row[0] for row in cur.fetchall()]

# SQLAlchemy (any backend)
from sqlalchemy import inspect
live_tables = inspect(engine).get_table_names()

# Static config (YAML, JSON, etc.)
import json
with open("schema.json") as f:
    live_tables = json.load(f)["tables"]

Then pass the list to any validator:

validator = SchemaValidator(live_tables)
success, msg = validator.validate("SELECT * FROM users JOIN orders ON ...")

Integration Examples

Detailed guides with runnable scripts for specific database integrations:

Integration	Guide
AWS Glue Catalog + Athena	examples/glue_athena/
GCP BigQuery	examples/bigquery/
Azure Synapse Analytics	examples/azure_synapse/
AWS Redshift (Data API)	examples/redshift/
PostgreSQL	examples/postgresql/
MySQL	examples/mysql/

API Reference

validate_query(sql_query, live_tables, *, dialect=None)

Standalone function for one-off validation.

Parameter	Type	Description
sql_query	str	The SQL query to validate
live_tables	list[str]	Tables that exist in the schema
dialect	str | None	SQL dialect ("postgres", "mysql", etc.)

Returns: tuple[bool, str]

SchemaValidator(live_tables, *, case_sensitive=False, preserve_schema=False)

Class-based validator with pre-computed table lookups.

Option	Default	Description
case_sensitive	False	Enable case-sensitive table name matching
preserve_schema	False	Match full schema.table names instead of base name

Methods:

Method	Description
validate(sql_query, dialect)	Validate a query against the schema
update_schema(live_tables)	Hot-swap the schema at runtime
table_exists(table_name)	Check if a specific table exists
get_table_count()	Return the number of registered tables

CachedSchemaValidator

Extends SchemaValidator with LRU caching. Accepts an additional cache_size parameter (default: 128).

Additional methods: clear_cache(), get_cache_info()

Project Structure

sqldrift/
├── pyproject.toml
├── LICENSE
├── README.md
├── src/
│   └── sqldrift/
│       ├── __init__.py        # Public API
│       ├── validator.py       # Core validate_query function
│       └── optimized.py       # SchemaValidator & CachedSchemaValidator
├── tests/
│   └── test_validator.py      # pytest suite (20 tests)
├── examples/
│   └── usage_examples.py
└── benchmarks/
    └── benchmark.py

Performance

Benchmarked with 4,000 tables:

Method	Speedup vs function
SchemaValidator	~2.6x faster
CachedValidator	~282x faster

python benchmarks/benchmark.py

Testing

pip install -e ".[dev]"
pytest

License

MIT