Gibt es eine schöne Möglichkeit, DML in SQLAlchemy zu deaktivieren?

Question

Ich möchte meinem Skript eine --dry-run-Option hinzufügen, um zu verhindern, dass Updates an der Datenbank ausgegeben werden. Ich möchte, dass es stattdessen etwas protokolliert (z. B. die Anweisung, die es ausführen würde).

Gibt es einen schönen Weg, um es in SQLAlchemy zu erreichen? Z.B. Gibt es etwas gut dokumentiert, um/affe-patch in einem Session Objekt zu wechseln?

Answer 1

Hier ist ein einfaches Beispielskript, joins the session to an existing transactional connection, die zurückgesetzt wird, wenn der richtige Schalter angegeben wird.

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())) 

Probelauf:

% 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