python数据库orm之sqlalchemy的excute执行方式
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()使用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()