execute()的三种方式
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine, Table, MetaData, Column, Integer, String
def sqlalchemy_test():
engine = create_engine("mysql+pymysql://username:password@localhost:3306/database_name?charset=utf8", echo=True)
meta = MetaData()
test_table = Table("tb_test", meta,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("name", String(255))
)
# 方式1
res_1 = engine.execute(test_table.select())
for re in res_1:
# dosomething
res_1.close()
# 方式2
conn = engine.connect()
res_2 = conn.execute(test_table.select())
for re in res_2:
# dosomething
res_2.close()
# 方式3
session_db = sessionmaker(bind=engine)
session = session_db()
res_3 = session.execute(test_table.select())
for re in res_3:
# dosomething
session.close()
这里需要说明一下,指定数据库编码的时候,经测试engine自带的设置编码格式的参数encoding没有效果(不知道是不是只有mysql+pymysql有这样的情况),所以使用?charset=utf8的形式指定数据库编码
三种方式的区别
感觉engine.execute()和connection.execute()没有太大的区别,engine.execute()会转化成connection.execute()。
Engine
object (instantiated via create_engine()
)
->
Connection
object (instantiated via engine_instance.connect()
)
->
connection.execute({*SQL expression*})
session.execute()最终也是使用connection.execute(),而且使用session对象与数据库交互是orm推荐的方式。感觉session是有事物的概念的,session.add(), session.commit(), session.close(),session.rollback()。既然官方推荐这种方式,以后还是尽量使用session来与数据库交互吧。
另外,使用try,except捕获异常,有回滚操作的时候,建议使用以下方式:
class ThingOne(object):
def go(self, session):
session.query(FooBar).update({"x": 5})
class ThingTwo(object):
def go(self, session):
session.query(Widget).update({"q": 18})
def run_my_program():
session = Session()
try:
ThingOne().go(session)
ThingTwo().go(session)
session.commit()
except:
session.rollback()
raise
finally:
session.close()
原文链接:https://blog.csdn.net/qq_25730711/article/details/53407459
网友评论