API Reference

Classifier

class sql_query_tagger.SQLClassifier(engine, version, max_query_length=10000, analysis_cache_size=2048)

Bases: object

Static SQL query classifier with engine-aware security analysis.

Usage:

classifier = SQLClassifier(engine=”postgresql”, version=”16”) result = classifier.classify_query(“SELECT * FROM users”)

Parameters:

• engine (str)

• version (str)

• max_query_length (int)

• analysis_cache_size (int)

clean_sql_string(sql_query)

Strip comments and normalize whitespace in a raw SQL string.

Parameters:

sql_query (str)

Return type:

str

classify_query(sql_query)

Classify a single SQL query string. Raises ValueError on invalid input.

Parameters:

sql_query (str)

Return type:

ClassificationResult

Result types

class sql_query_tagger.ClassificationResult(query_type, security_analysis, cleaned_query, parsing_success, processing_time_ms, engine, engine_version, metadata=<factory>)

Bases: object

Complete classification result for a single query.

Parameters:

• query_type (QueryType)

• security_analysis (SecurityAnalysis)

• cleaned_query (str)

• parsing_success (bool)

• processing_time_ms (float)

• engine (str)

• engine_version (str)

• metadata (Dict[str, str | int | float])

query_type: QueryType

security_analysis: SecurityAnalysis

cleaned_query: str

parsing_success: bool

processing_time_ms: float

engine: str

engine_version: str

metadata: Dict[str, str | int | float]

class sql_query_tagger.SecurityAnalysis(is_suspicious=False, risk_level=RiskLevel.LOW, confidence_score=0.0, detected_patterns=<factory>, anomaly_indicators=<factory>, engine_specific_risks=<factory>, recommendation='')

Bases: object

Security analysis results for a single query.

Parameters:

• is_suspicious (bool)

• risk_level (RiskLevel)

• confidence_score (float)

• detected_patterns (List[str])

• anomaly_indicators (List[str])

• engine_specific_risks (List[str])

• recommendation (str)

is_suspicious: bool = False

risk_level: RiskLevel = 'LOW'

confidence_score: float = 0.0

detected_patterns: List[str]

anomaly_indicators: List[str]

engine_specific_risks: List[str]

recommendation: str = ''

class sql_query_tagger.QueryType(*values)

Bases: Enum

SQL query classification categories.

DDL = 'DDL'

DML = 'DML'

DQL = 'DQL'

DCL = 'DCL'

TCL = 'TCL'

UTILITY = 'UTILITY'

PROCEDURAL = 'PROCEDURAL'

ADMIN = 'ADMIN'

UNKNOWN = 'UNKNOWN'

class sql_query_tagger.RiskLevel(*values)

Bases: Enum

Security risk levels.

LOW = 'LOW'

MEDIUM = 'MEDIUM'

HIGH = 'HIGH'

CRITICAL = 'CRITICAL'

Exceptions

exception sql_query_tagger.UnsupportedEngineError

Bases: ValueError

Raised when a requested RDS engine name doesn’t match any known EngineProfile.

Engine registry

sql_query_tagger.engines.registry.get_profile(engine, version)

Resolve an RDS engine name + version to an EngineProfile instance.

Raises UnsupportedEngineError if the engine name is unknown. An unknown version for a known engine is not an error - RDS adds versions faster than this package can track them, so it warns and falls back to the engine’s base patterns with no version-specific overrides.

Parameters:

• engine (str)

• version (str)

Return type:

EngineProfile

class sql_query_tagger.engines.base.EngineProfile(version)

Bases: ABC

Base class for RDS engine-specific classification data.

Subclasses set class-level attributes for the engine’s base pattern set, then optionally override version_overrides to layer in version-specific differences (e.g. functions/catalogs added in a later major version).

Parameters:

version (str)

name: str = ''

dialect_name: str = ''

supported_versions: List[str] = []

utility_commands: Dict[str, str] = {}

admin_commands: Dict[str, str] = {}

procedural_commands: Dict[str, str] = {}

tcl_commands: Dict[str, str] = {}

dcl_commands: Dict[str, str] = {}

injection_patterns: Dict[str, List[str]] = {}

risk_weights: Dict[str, float] = {}

dangerous_functions: Set[str] = {}

sensitive_catalogs: Set[str] = {}

version_overrides(version)

Return version-specific overrides to merge over base class attributes.

Default: no overrides. Subclasses override this for engines/versions where syntax, functions, or catalogs differ meaningfully by version.

Parameters:

version (str)

Return type:

dict

command_tables()

Merge all keyword->QueryType heuristic tables into one lookup.

Return type:

Dict[str, str]