SQLAlchemyでDMLを無効にする良い方法はありますか？

Question

--dry-runオプションをスクリプトに追加して、データベースへの更新を発行しないようにします。代わりに何かをログに記録したいと思います（たとえば、実行する文）。

SQLAlchemyでそれを実現する良い方法はありますか？例えば。 Sessionオブジェクトの中に/ monkey-patchを切り替えることについてよく文書化されていますか？

Answer 1

正しいスイッチが指定されている場合は、ロールバックされるjoins the session to an existing transactional connectionの簡単なスクリプト例を示します。

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

サンプル実行：

% 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