Backstop

SQLAlchemy

Attach Backstop protection to an existing SQLAlchemy engine with the Python SDK.

SQLAlchemy is widely used in Python data pipelines, web backends, and AI agent stacks. Backstop's current Python SDK integration attaches to an existing SQLAlchemy engine and intercepts statements through SQLAlchemy's before_cursor_execute hook.

Engine hook integration

This integration keeps your existing engine and session usage intact. Backstop classifies the SQL before the DBAPI cursor executes, and in Python SDK protect mode it can capture before-images for table-level recoverable operations.

from sqlalchemy import create_engine
import backstop

engine = create_engine("postgresql://postgres:password@localhost:5432/mydb")
safe_engine = backstop.protect_engine(
    engine=engine,
    storage="s3://prod-snapshots@http://localhost:9000",
    actor="sqlalchemy-pipeline",
    mode="protect",
)

# Use safe_engine anywhere you'd use engine
with safe_engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users LIMIT 10"))

ORM usage

The proxy is transparent to SQLAlchemy's ORM:

from sqlalchemy.orm import Session
from your_models import User

with Session(safe_engine) as session:
    # SELECT — risk_level: SAFE, executes immediately
    users = session.query(User).filter(User.active == True).all()

    # DELETE — risk_level: HIGH or CRITICAL depending on scope
    session.query(User).filter(User.created_at < cutoff).delete()
    session.commit()  # Backstop intercepts at flush time

Alembic migrations

Route migration SQL through an engine that has protect_engine(...) attached if you want local logging and snapshot behavior:

# alembic/env.py
from sqlalchemy import engine_from_config

def run_migrations_online():
    connectable = engine_from_config(config.get_section(config.config_ini_section))
    safe_connectable = backstop.protect_engine(
        engine=connectable,
        storage=os.environ["BACKSTOP_STORAGE_URL"],
        actor="alembic-migrations",
        mode="monitor",
    )

    with safe_connectable.connect() as connection:
        context.configure(connection=connection)
        with context.begin_transaction():
            context.run_migrations()

For production schema changes that need approval and stronger safety policy, prefer routing through the gateway instead of relying only on the local Python SDK hook.

Behavior notes

The SQLAlchemy integration follows the Python SDK behavior:

engine = backstop.protect_engine(
    engine=engine,
    storage="s3://prod-snapshots",
    actor="sqlalchemy-app",
    mode="block",
)
  • protect snapshots then executes supported destructive operations
  • monitor logs/classifies but does not block
  • block raises on CRITICAL local classifications
  • snapshot failure in protect mode is logged and the original query still executes
  • unrecoverable operations such as DROP DATABASE should be handled at the gateway or infrastructure layer

Configuration reference

ParameterTypeDefaultDescription
engineSQLAlchemy EnginerequiredExisting engine to instrument
storagestringrequiredS3-compatible storage URL for snapshots
actorstringNoneIdentifier recorded in local audit logs
modestring"protect"One of protect, monitor, or block