我的数据库模块中有一个(匿名)函数:
def fix_publishing_dates(row_id, last_time=None, next_time=None,
next_index=1, user="python_script"):
sql = """
UPDATE
schema.table
SET
last_time = :last_time
, next_time = :next_time
, next_index = :next_index
, col4 = SYSDATE
, col5 = :user_id
, is_active = 1
WHERE
id = :row_id
"""
with closing(Session()) as s:
with s.begin_nested():
user_id = get_userid_by_name(user)
args = dict(
last_time=last_time,
next_time=next_time,
next_index=next_index,
row_id=row_id,
user_id=user_id,
)
s.execute(sql, args)
s.flush()
s.commit()
由于某种原因,这是行不通的。我在上表中查询is_active=1,得到零行。我在这里做了什么明显的错误吗?在
我不想使用SQLAlchemy ORM并为此添加大量的样板表类*;我只喜欢将Session()与文本查询一起用于事务支持。在
*:也不能通过自省来降低我的启动时间;到这个数据库的网络管道很慢。在
此代码(匿名方式不同)也不起作用:
def fix_publishing_dates(**kwargs):
sql = insert_query_here
user_id = get_userid_by_name(user)
args = dict(kwargs)
print "*" * 50
print "* About to update database with values: {}".format(args)
print "*" * 50
result = engine.execute(sql, args)
print "Row count is:", result.rowcount
#import ipdb;ipdb.set_trace()
#s.commit()
我不认为会话在不使用ORM对象的情况下无法管理事务。但您可以直接使用引擎上的事务:
相关问题 更多 >
编程相关推荐