有沒有一種很好的方法來禁用SQLAlchemy中的DML？

Question

我想爲我的腳本添加一個--dry-run選項，以防止它向數據庫發出任何更新。我希望它記錄一些東西（例如它將運行的語句）。

有沒有一種很好的方法在SQLAlchemy中實現它？例如。在Session對象中切換/猴子補丁有什麼好記錄嗎？

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