CODEWITHSUNDEEP
pythonsqlalchemy
9000
9000

Reputation: 40894

Is there a nice way to disable DML in SQLAlchemy?

I want to add a --dry-run option to my script, to prevent it from issuing any updates to the database. I'd like it to log something instead (e.g. the statement it would run).

Is there a nice way to achieve it in SQLAlchemy? E.g. is there something well-documented to switch / monkey-patch in a Session object?

Upvotes: 3

Views: 928

Answers (1)

Ilja Everilä
Ilja Everilä

Reputation: 52937

Here's a simple example script that joins the session to an existing transactional connection, which is rolled back, if given the right switch.

import argparse
from contextlib import contextmanager
from sqlalchemy import create_engine, table, column
from sqlalchemy.orm import sessionmaker
import logging

_Session = sessionmaker()
logging.basicConfig()


@contextmanager
def _destructive_run(engine):
    session = _Session(bind=engine)
    try:
        yield session

    finally:
        session.close()


@contextmanager
def _dry_run(engine):
    logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
    conn = engine.connect()
    trans = conn.begin()
    session = _Session(bind=conn)
    try:
        yield session

    finally:
        session.close()
        trans.rollback()
        conn.close()

_args = argparse.ArgumentParser("Super Dry Runner")
_args.add_argument("--dry-run", dest='context', action='store_const',
                   const=_dry_run, default=_destructive_run)
_args.add_argument("engine", metavar="DB_URL", type=create_engine)
_args.add_argument("values", metavar="VALUE", nargs="+",
                   type=lambda v: { "value": v })


def main(*, engine, context, values):
    with context(engine) as session:
        table1 = table("table1", column("value"))
        session.execute(table1.insert().values(values))
        session.commit()

if __name__ == '__main__':
    main(**vars(_args.parse_args()))

Sample run:

 % sqlite3 test.db "create table table1 (value text);"
 % python dry.py sqlite:///test.db test1 test2 test3            
 % sqlite3 test.db "select * from table1;"            
test1
test2
test3
 % python dry.py --dry-run sqlite:///test.db test4 test5 test6
INFO:sqlalchemy.engine.base.Engine:SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
INFO:sqlalchemy.engine.base.Engine:()
INFO:sqlalchemy.engine.base.Engine:SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
INFO:sqlalchemy.engine.base.Engine:()
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:INSERT INTO table1 (value) VALUES (?), (?), (?)
INFO:sqlalchemy.engine.base.Engine:('test4', 'test5', 'test6')
INFO:sqlalchemy.engine.base.Engine:ROLLBACK
 % sqlite3 test.db "select * from table1;"                    
test1
test2
test3

Upvotes: 4

Related Questions