SQLAlchemy(无ORM):update语句未提交

2024-03-29 14:41:36 发布

您现在位置:Python中文网/ 问答频道 /正文

我的数据库模块中有一个(匿名)函数:

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()与文本查询一起用于事务支持。在

*:也不能通过自省来降低我的启动时间;到这个数据库的网络管道很慢。在

编辑1

  • 我使用的是一个Oracle 11数据库,它是通过cx Oracle。在
  • 如果重要的话,其中一个绑定值有时是None/
  • 此代码(匿名方式不同)也不起作用:

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

Tags: noneid数据库sqlindextimeisdef
1条回答
网友
1楼 · 发布于 2024-03-29 14:41:36

我不认为会话在不使用ORM对象的情况下无法管理事务。但您可以直接使用引擎上的事务:

with engine.begin() as conn:
    conn.execute(...)

相关问题 更多 >