Safety API

SQL query guard service.

class sqldbagent.safety.guard.QueryGuardService(policy, dialect)

Bases: object

Guard SQL through AST inspection and normalization.

Parameters:

• policy (SafetySettings)

• dialect (Dialect)

__init__(policy, dialect)

Initialize the query guard.

Parameters:

• policy (SafetySettings) – Safety policy settings.

• dialect (Dialect) – Datasource dialect.

Return type:

None

lint(sql)

Parse and normalize SQL without applying guard rewrites.

Parameters:

sql (str) – SQL text to lint.

Returns:

Lint result.

Return type:

QueryGuardResult

guard(sql, *, max_rows=None)

Parse, validate, and normalize SQL under the active policy.

Parameters:

• sql (str) – SQL text to guard.

• max_rows (int | None, default: None) – Optional row-limit override for this evaluation.

Returns:

Guard result.

Return type:

QueryGuardResult

Guarded sync and async SQL execution services.

class sqldbagent.safety.execution.SafeQueryService(*, engine, guard, async_engine=None)

Bases: object

Execute read-only SQL only after it passes the guard layer.

Parameters:

• engine (Engine)

• guard (QueryGuardService)

• async_engine (AsyncEngine | None, default: None)

__init__(*, engine, guard, async_engine=None)

Initialize the safe query service.

Parameters:

• engine (Engine) – Sync SQLAlchemy engine.

• guard (QueryGuardService) – Shared SQL guard service.

• async_engine (AsyncEngine | None, default: None) – Optional async SQLAlchemy engine.

Return type:

None

run(sql, *, max_rows=None)

Guard and execute SQL synchronously.

Parameters:

• sql (str) – SQL text to execute.

• max_rows (int | None, default: None) – Optional row-limit override.

Returns:

Guard and execution result.

Return type:

QueryExecutionResult

async run_async(sql, *, max_rows=None)

Guard and execute SQL asynchronously.

Parameters:

• sql (str) – SQL text to execute.

• max_rows (int | None, default: None) – Optional row-limit override.

Returns:

Guard and execution result.

Return type:

QueryExecutionResult

Raises:

ConfigurationError – If no async engine is configured.

Safety models.

class sqldbagent.safety.models.QueryGuardResult(**data)

Bases: BaseModel

Guard evaluation result.

Variables:

• allowed – Whether the query passed safety validation.

• statement_type – Root SQL statement type.

• dialect – SQL dialect used for parsing and normalization.

• original_sql – Original SQL text.

• normalized_sql – Normalized SQL text after linting or guarding.

• row_limit_applied – Whether the guard injected or reduced a row limit.

• max_rows – Maximum rows allowed by policy.

• referenced_schemas – Schemas referenced by the statement.

• referenced_tables – Tables referenced by the statement.

• reasons – Validation failure reasons when not allowed.

• summary – Generated short summary.

Parameters:

• data (Any)

• allowed (bool)

• statement_type (str | None)

• dialect (str)

• original_sql (str)

• normalized_sql (str | None)

• row_limit_applied (bool)

• max_rows (int | None)

• referenced_schemas (list[str])

• referenced_tables (list[str])

• reasons (list[str])

• summary (str | None)

allowed: bool

statement_type: str | None

dialect: str

original_sql: str

normalized_sql: str | None

row_limit_applied: bool

max_rows: int | None

referenced_schemas: list[str]

referenced_tables: list[str]

reasons: list[str]

summary: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].