sql-query-tagger

Static SQL query classifier and injection-risk analyzer for all Amazon RDS database engines.

sql-query-tagger parses a SQL query string and tells you:

• Query type: DDL, DML, DQL, DCL, TCL, UTILITY, PROCEDURAL, ADMIN, or UNKNOWN
• Security risk: LOW / MEDIUM / HIGH / CRITICAL, with the specific patterns that triggered it (injection patterns, stacked queries, destructive DDL, engine-specific dangerous functions/catalogs)

No database connection required — this is purely static analysis over the query text.

Supported engines

Listed versions are the ones with version-specific pattern tuning. Other versions of a supported engine still work — they fall back to the engine's base pattern set with a warning logged.

Install

pip install sql-query-tagger

Quickstart

from sql_query_tagger import SQLClassifier

classifier = SQLClassifier(engine="postgresql", version="16")
result = classifier.classify_query("SELECT * FROM users WHERE id = 1 OR 1=1")

print(result.query_type)                       # QueryType.DQL
print(result.security_analysis.risk_level)     # RiskLevel.HIGH
print(result.security_analysis.detected_patterns)
print(result.security_analysis.recommendation)

More examples

Blocking a destructive statement

from sql_query_tagger import SQLClassifier

classifier = SQLClassifier(engine="mysql", version="8.0")
result = classifier.classify_query("SELECT 1; DROP TABLE users;")

print(result.security_analysis.risk_level)         # RiskLevel.CRITICAL
print(result.security_analysis.is_suspicious)      # True
print(result.security_analysis.detected_patterns)
# ['stacked_queries_multiple_statements', 'stacked_queries_dangerous_followup', ...]

Engine-specific dangerous function

classifier = SQLClassifier(engine="sqlserver", version="2022")
result = classifier.classify_query("EXEC xp_cmdshell 'dir'")

print(result.security_analysis.engine_specific_risks)
# ['dangerous_function_xp_cmdshell']

Classifying query type only (DDL/DML/DQL/...)

classifier = SQLClassifier(engine="postgresql", version="16")

for sql in ["CREATE TABLE t (id INT)", "INSERT INTO t VALUES (1)", "SELECT * FROM t"]:
    result = classifier.classify_query(sql)
    print(sql, "->", result.query_type)
# CREATE TABLE t (id INT) -> QueryType.DDL
# INSERT INTO t VALUES (1) -> QueryType.DML
# SELECT * FROM t -> QueryType.DQL

Handling invalid input and unknown engines

from sql_query_tagger import SQLClassifier, UnsupportedEngineError

try:
    classifier = SQLClassifier(engine="db2", version="11.5")
except UnsupportedEngineError as e:
    print(f"Unsupported engine: {e}")

classifier = SQLClassifier(engine="postgresql", version="16")
try:
    classifier.classify_query("")
except ValueError as e:
    print(f"Invalid query: {e}")

Performance

classify_query memoizes its security analysis and query-type classification per SQLClassifier instance, keyed by the cleaned query text (this is the part that calls into sqlglot and runs the regex scans). Real traffic tends to repeat the same templated query shapes (an ORM or app issuing the same statement with different parameter values folded out), so this cache turns repeats into a dict lookup instead of a re-parse. Tune or disable it with SQLClassifier(..., analysis_cache_size=N) (0 disables caching).

Run the benchmark yourself:

python benchmarks/bench_classify.py

Measured on a single thread, 10,000 queries per scenario:

The unique-query case is the floor — it's bound by sqlglot's parser, which accounts for roughly 75% of per-call time (confirmed via cProfile). Since correct query-type classification depends on that parse, this floor isn't something sql-query-tagger can optimize away without giving up accuracy; the cache is what makes high-repeat production traffic fast.

Testing

pip install -e ".[dev]"
pytest --cov=sql_query_tagger --cov-report=term-missing

Test report

The suite has 114 tests across the classifier pipeline, every engine profile, the registry, and the public package API, with 99% statement coverage:

Name                                      Stmts   Miss  Cover   Missing
-----------------------------------------------------------------------
sql_query_tagger\__init__.py                        4      0   100%
sql_query_tagger\classifier.py                    132      1    99%   188
sql_query_tagger\engines\__init__.py                0      0   100%
sql_query_tagger\engines\aurora_mysql.py            6      0   100%
sql_query_tagger\engines\aurora_postgresql.py       6      0   100%
sql_query_tagger\engines\base.py                   32      0   100%
sql_query_tagger\engines\mariadb.py                 8      0   100%
sql_query_tagger\engines\mysql.py                  19      0   100%
sql_query_tagger\engines\oracle.py                 15      0   100%
sql_query_tagger\engines\postgresql.py             15      0   100%
sql_query_tagger\engines\registry.py               25      0   100%
sql_query_tagger\engines\sqlserver.py              15      0   100%
sql_query_tagger\exceptions.py                      1      0   100%
sql_query_tagger\types.py                          30      0   100%
-----------------------------------------------------------------------
TOTAL                                       308      1    99%
114 passed in 0.54s

The one uncovered line is a defensive fallback for a CTE-only top-level parse shape that sqlglot does not currently produce in practice.

Limitations

• Static analysis only — it inspects query text, not a live schema, so it cannot tell you whether referenced tables/columns exist.
• The stacked-query check splits on ; without parsing string literals, so a benign query containing a semicolon inside a string (e.g. 'a; b') may be flagged as multiple statements. Treat the risk score as a signal, not a verdict.

License

Apache License 2.0