proxql 0.1.0

SQL validation library that blocks destructive queries from LLM-generated SQL

ProxQL

A 50-line firewall to stop your AI from dropping tables

Installation • Quick Start • Modes • API Reference • Integrations

---

The Problem

You're building an AI agent that talks to your database. But what happens when:

• 🔥 Your LLM hallucinates and runs DROP TABLE users
• 🔓 It queries SELECT * FROM employees and leaks salaries
• 💸 It writes a cartesian join that scans 10 billion rows

ProxQL validates every query before it touches your data.

Installation

pip install proxql

Quick Start

import proxql

# ✓ Safe queries pass
proxql.validate("SELECT * FROM users").is_safe  # True
proxql.is_safe("SELECT * FROM products")        # True

# ✗ Dangerous queries are blocked
result = proxql.validate("DROP TABLE users")
result.is_safe   # False
result.reason    # "Statement type 'DROP' is not allowed in read_only mode"

# ✗ Unauthorized tables are blocked
result = proxql.validate(
    "SELECT * FROM employees",
    allowed_tables=["products", "categories"]
)
result.is_safe   # False
result.reason    # "Table 'employees' is not in allowed tables list"

Modes

Mode	Allowed Statements	Use Case
read_only	SELECT only	Analytics, reporting, read-only agents
write_safe	SELECT, INSERT, UPDATE	CRUD operations (no destructive ops)
custom	You define	Full control over allowed/blocked statements

Read-Only Mode (Default)

import proxql

# Only SELECT statements pass
proxql.is_safe("SELECT * FROM users")           # True
proxql.is_safe("INSERT INTO logs VALUES (1)")   # False
proxql.is_safe("DELETE FROM users")             # False
proxql.is_safe("DROP TABLE users")              # False

Write-Safe Mode

from proxql import Validator

validator = Validator(mode="write_safe")

validator.validate("SELECT * FROM users").is_safe    # True
validator.validate("INSERT INTO users ...").is_safe  # True
validator.validate("UPDATE users SET ...").is_safe   # True
validator.validate("DELETE FROM users").is_safe      # False  (blocked)
validator.validate("DROP TABLE users").is_safe       # False  (blocked)
validator.validate("TRUNCATE TABLE users").is_safe   # False  (blocked)

Custom Mode

from proxql import Validator

# Allow only specific statements
validator = Validator(
    mode="custom",
    allowed_statements=["SELECT", "INSERT"],
)
validator.validate("SELECT * FROM users").is_safe  # True
validator.validate("INSERT INTO logs ...").is_safe # True
validator.validate("UPDATE users SET ...").is_safe # False

# Or block specific statements
validator = Validator(
    mode="custom",
    blocked_statements=["DROP", "TRUNCATE"],
)
validator.validate("SELECT * FROM users").is_safe  # True
validator.validate("DROP TABLE users").is_safe     # False

Table Allowlist

Restrict queries to specific tables:

from proxql import Validator

validator = Validator(
    mode="read_only",
    allowed_tables=["products", "categories", "reviews"]
)

validator.validate("SELECT * FROM products").is_safe      # True
validator.validate("SELECT * FROM employees").is_safe     # False

# Also detects tables in subqueries, CTEs, and JOINs
validator.validate("""
    SELECT * FROM (SELECT * FROM secret_table) AS t
""").is_safe  # False - secret_table detected in subquery

SQL Dialect Support

ProxQL uses sqlglot under the hood, supporting 20+ SQL dialects:

from proxql import Validator

# PostgreSQL
pg_validator = Validator(mode="read_only", dialect="postgres")
pg_validator.validate("SELECT * FROM users LIMIT 10 OFFSET 5")

# MySQL
mysql_validator = Validator(mode="read_only", dialect="mysql")
mysql_validator.validate("SELECT * FROM users LIMIT 5, 10")

# Snowflake, BigQuery, DuckDB, etc.

Supported dialects: postgres, mysql, sqlite, snowflake, bigquery, redshift, duckdb, presto, trino, spark, and more.

API Reference

proxql.validate(sql, *, mode, allowed_tables, dialect)

Validate a SQL query string.

proxql.validate(
    sql: str,                              # The SQL query to validate
    *,
    mode: str = "read_only",               # "read_only" | "write_safe" | "custom"
    allowed_tables: list[str] | None = None,  # Optional table whitelist
    dialect: str | None = None,            # SQL dialect (auto-detected if None)
) -> ValidationResult

proxql.is_safe(sql, **kwargs)

Convenience wrapper that returns just the boolean result.

proxql.is_safe("SELECT * FROM users")  # True
proxql.is_safe("DROP TABLE users")     # False

proxql.Validator

For repeated validations, create a Validator instance:

from proxql import Validator

validator = Validator(
    mode: str = "read_only",               # Validation mode
    allowed_tables: list[str] | None = None,  # Table whitelist
    allowed_statements: list[str] | None = None,  # For custom mode
    blocked_statements: list[str] | None = None,  # For custom mode
    dialect: str | None = None,            # SQL dialect
)

result = validator.validate(sql: str) -> ValidationResult

ValidationResult

@dataclass(frozen=True)
class ValidationResult:
    is_safe: bool                    # Whether the query passed validation
    reason: str | None = None        # Explanation if blocked
    statement_type: str | None = None  # SELECT, INSERT, DROP, etc.
    tables: list[str] = []           # Tables referenced in query

    def __bool__(self) -> bool:      # Can use in boolean context
        return self.is_safe

Integrations

LangChain

from langchain_community.utilities import SQLDatabase
from proxql import Validator

db = SQLDatabase.from_uri("postgresql://localhost/mydb")
validator = Validator(mode="read_only")

def safe_query(query: str) -> str:
    result = validator.validate(query)
    if not result.is_safe:
        raise ValueError(f"Query blocked: {result.reason}")
    return db.run(query)

# Use safe_query instead of db.run in your agent

Raw Database Drivers

import psycopg2
from proxql import Validator

conn = psycopg2.connect("...")
validator = Validator(mode="read_only", allowed_tables=["products"])

def execute_safe(cursor, query: str):
    result = validator.validate(query)
    if not result.is_safe:
        raise ValueError(f"Blocked: {result.reason}")
    return cursor.execute(query)

FastAPI Middleware

from fastapi import FastAPI, HTTPException
from proxql import Validator

app = FastAPI()
validator = Validator(mode="read_only")

@app.post("/query")
async def run_query(query: str):
    result = validator.validate(query)
    if not result.is_safe:
        raise HTTPException(400, f"Query blocked: {result.reason}")
    # Execute query...

Edge Cases Handled

ProxQL correctly detects:

• Subqueries: SELECT * FROM (SELECT * FROM secret_table) AS t
• CTEs: WITH temp AS (SELECT * FROM secret) SELECT * FROM temp
• JOINs: SELECT * FROM a JOIN b ON ... — checks all tables
• Multi-statement: SELECT 1; DROP TABLE users; — blocks if any unsafe
• Comments: SELECT * /* DROP TABLE */ FROM users — comments ignored
• Case sensitivity: drop TABLE Users normalized correctly

Why ProxQL?

"You wouldn't give a junior intern root access to production. Why are you giving it to a hallucinating AI?"

Every AI framework (LangChain, CrewAI, AutoGen) lets you connect to databases. None of them protect you from what the AI might do once connected.

ProxQL is the missing safety layer.

Contributing

git clone https://github.com/zeredbaron/proxql.git
cd proxql
pip install -e ".[dev]"
pytest

License

Apache License 2.0 — See LICENSE for details.

---

Built for the agentic future 🤖