sqlalchemy 8 会话与异常

先创建一个c8数据库。c8.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///c8.db')    #引擎
Session = sessionmaker(bind=engine)    #会话
session = Session()from sqlalchemy import Table, Column, Integer, Numeric, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref
Base = declarative_base()class Cookie(Base):__tablename__ = 'cookies'cookie_id = Column(Integer(), primary_key=True)cookie_name = Column(String(30), index=True)cookie_recipe_url = Column(String(255))cookie_sku = Column(String(55))quantity = Column(Integer())unit_cost = Column(Numeric(12,2))def __init__(self, name, recipe_url=None, sku=None, quantity=0, unit_cost=0.00):self.cookie_name = nameself.cookie_recipe_url = recipe_urlself.cookie_sku = skuself.quantity = quantityself.unit_cost = unit_costdef __repr__(self):return "Cookie(cookie_name='{self.cookie_name}',"\"cookie_recipe_url='{self.cookie_recipe_url}',"\"cookie_sku='{self.cookie_sku}',"\"quantity={self.quantity},"\"unit_cost={self.unit_cost})".format(self=self)class User(Base):__tablename__ = 'users'user_id = Column(Integer(), primary_key=True)username = Column(String(15), nullable=False, unique=True)email_address = Column(String(255), nullable=False)phone = Column(String(20), nullable=False)password = Column(String(25), nullable=False)created_on = Column(DateTime(), default=datetime.now)updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)def __init__(self, username, email_address, phone, password):self.username = usernameself.email_address = email_addressself.phone = phoneself.password = passworddef __repr__(self):return "User(username='{self.username}', "\"email_address='{self.email_address}', "\"phone='{self.phone}', "\"password='{self.password}')".format(self=self)class Order(Base):__tablename__ = 'orders'order_id = Column(Integer(), primary_key=True)user_id = Column(Integer(), ForeignKey('users.user_id'))shipped = Column(Boolean(), default=False)user = relationship('User', backref=backref('orders', order_by=order_id))def __repr__(self):return "Order(user_id={self.user_id}, " \"shipped={self.shipped})".format(self=self)class LineItems(Base):__tablename__ = 'line_items'line_items_id = Column(Integer(), primary_key=True)order_id = Column(Integer(), ForeignKey('orders.order_id'))cookie_id = Column(Integer(), ForeignKey('cookies.cookie_id'))quantity = Column(Integer())extended_cost = Column(Numeric(12,2))order = relationship('Order', backref=backref('line_items', order_by=line_items_id))cookie = relationship('Cookie', uselist=False)def __repr__(self):return "LineItems(order_id={self.order_id}, " \"cookie_id={self.cookie_id}, " \"quantity={self.quantity}, " \"extended_cost={self.extended_cost})".format(self=self)Base.metadata.create_all(engine)    #表持久化

在查询一个对象时，对象会在会话的几个状态间切换。

transient 瞬时状态：实例不在会话中，也不在数据库中。（不在会话也不在数据库）

pending 挂起状态：实例有add加入到会话中，但未刷新或提交（在会话中不在数据库中）

persistent 持久化状态：会话中的对象在数据库中有对应的记录。（在会话中也在数据库中）flush和commit以后，都是这个状态

detached 脱管状态：实例不再与会话相连，但在数据库中有记录（不在会话中在数据库中）

查看实例的状态用inspect()。 c8_insert.py

from c8 import Cookie, session
c = Cookie('chocolate chip','http://some.aweso.me/cookie/recipe.html','cc01',12,0.5)    #新建一个实例
from sqlalchemy import inspect
i = inspect(c)
for state in ['transient','pending','persistent','detached']:print('{:>10}:{}'.format(state, getattr(i, state)))    #用getattr遍历

 transient:Truepending:False
persistent:Falsedetached:False

add

#...
session.add(c)
from sqlalchemy import inspect
i = inspect(c)
for state in ['transient','pending','persistent','detached']:print('{:>10}:{}'.format(state, getattr(i, state)))

 transient:Falsepending:True
persistent:Falsedetached:False

commit 和 flush

#...
session.add(c)
session.commit()
from sqlalchemy import inspect
i = inspect(c)
for state in ['transient','pending','persistent','detached']:print('{:>10}:{}'.format(state, getattr(i, state)))

#...
session.add(c)
session.flush()
from sqlalchemy import inspect
i = inspect(c)
for state in ['transient','pending','persistent','detached']:print('{:>10}:{}'.format(state, getattr(i, state)))

 transient:Falsepending:False
persistent:Truedetached:False

expunge()。让实例进入脱管状态。常用于把数据从一个会话移到另一个会话时，比如把主要数据库中的数据归档或合并到数据仓库中。

#。。。
session.add(c)
session.commit()
session.expunge(c)
from sqlalchemy import inspect
i = inspect(c)
for state in ['transient','pending','persistent','detached']:print('{:>10}:{}'.format(state, getattr(i, state)))

 transient:Falsepending:False
persistent:Falsedetached:True

除了用getattr()遍历，也可以用i.transient，i.pending等直接读出值。

from c8 import Cookie, session
c = Cookie('chocolate chip','http://some.aweso.me/cookie/recipe.html','cc01',12,0.5)
from sqlalchemy import inspect
i = inspect(c)    #从这步开始 inspect实例的状态
session.add(c)
session.commit()
session.expunge(c)
session.add(c)    #注意书上讲的“加回”到会话中，是接着上面的代码的。已经expunge了要重新add进会话
c.cookie_name = 'change chocolate chip'    #modify
#i = inspect(c)    #从这里开始inspect也行
print(i.modified)    #查看的是commit以后 的modified。如果光是add在会话里，没有commit过，那整个Cookie实例c都是modified的。for attr, attr_state in i.attrs.items():    #attr是个集合，有key和value，和正常集合一样用if attr_state.history.has_changes():print('{}: {}'.format(attr, attr_state.value))    #attr_state是个sqlalchemy.orm.state.AttributeState objectprint('history: {}\n'.format(attr_state.history))

输出True，inspect.modified可以看出在上一次的提交之后实例有没有发生变化。inspect.attr集合可以查看具体的变化。

回顾一下前面学的两个异常：

AttributeError：当试图访问一个不存在的属性时，就会出现AttributeError。比如要访问的列在ResultProxy中不存在。、

IntegrityError：当试图做一些违反列约束或表约束的事情时，就会出现IntegrityError。比如破坏唯一性约束。

MultipleResultFound：使用one()查询时，如果返回多个结果，就会报MultipleResultFound。

r = session.query(Cookie).one()    #之前已经commit过很多遍了，库里有很多cookie了，直接one就能触发

#报错
sqlalchemy.orm.exc.MultipleResultsFound: Multiple rows were found for one()

NoResultFound：使用one()查询时，没有返回任何结果。

异常处理 try/except

from sqlalchemy.orm.exc import MultipleResultsFound
try:r = session.query(Cookie).one()
except MultipleResultsFound as error:print('we found too many cookies.')

DetachedInstanceError：尝试访问实例的某些属性时，如果这个属性需要从数据库加载，但实例目前有没有连接到数据库，就会报错。范例，先创建一个用户cookiemon及其对应的order和lineitems

cookiemon = User('cookiemon','mon@cookie.com','111-111-1111','password')
session.add(cookiemon)    #add User
o1 = Order()
o1.user = cookiemon
session.add(o1)    #add Ordercc = session.query(Cookie).filter(Cookie.cookie_name == 'change chocolate chip').one()
line1 = LineItems(order=o1, cookie=cc, quantity=2, extended_cost=1.00)session.add(line1)    #add LineItems
session.commit()    #commit

order = session.query(Order).first()
session.expunge(order)
#session.add(order)    #若add回去，就正常。
print(order.line_items)
#line_items是order的一个relationship属性，relationship在默认情况下，除非特别请求，否则是不会在query时就加载出来的。
#像order的其他Column属性，order.order_id，order.user_id, order.user在query时就已经加载了。即便order被expunge了，也能print出来。
#line_items这个relationship缺少会话 来执行查询 重新加载，触发异常。

异常处理

from sqlalchemy.orm.exc import DetachedInstanceError
order = session.query(Order).first()
session.expunge(order)
try: print(order.line_items)
except DetachedInstanceError as error:session.add(order)    #处理办法就是 手动add回session。print(order.line_items)

类似DetachedInstanceError的异常还有：ObjectDeleteError、StaleDataError、ConcurrentModificationError，都是因为与实例、会话和数据库之间的信息不一致有关。

事务transaction，是一组语句。session在刚创建时，并没有连接到数据库。直到session执行第一个操作如query时，才会启动一个连接和一个事务。（即默认情况下，我们不用手动创建事务transaction）但当事务的一部分执行成功一部分执行失败时，触发了异常，我们要知道如何手动控制事务。

新事务环境，其实就是加了行CheckConstraint。 c8trans.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///c8trans.db')
Session = sessionmaker(bind=engine)
session = Session()
from sqlalchemy import Table, Column, Integer, Numeric, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref
from sqlalchemy import CheckConstraint
Base = declarative_base()class Cookie(Base):__tablename__ = 'cookies'__table_args__ = (CheckConstraint('quantity >= 0', name='quantity_positive'),)cookie_id = Column(Integer(), primary_key=True)cookie_name = Column(String(30), index=True)cookie_recipe_url = Column(String(255))cookie_sku = Column(String(55))quantity = Column(Integer())unit_cost = Column(Numeric(12,2))def __init__(self, name, recipe_url=None, sku=None, quantity=0, unit_cost=0.00):self.cookie_name = nameself.cookie_recipe_url = recipe_urlself.cookie_sku = skuself.quantity = quantityself.unit_cost = unit_costdef __repr__(self):return "Cookie(cookie_name='{self.cookie_name}',"\"cookie_recipe_url='{self.cookie_recipe_url}',"\"cookie_sku='{self.cookie_sku}',"\"quantity={self.quantity},"\"unit_cost={self.unit_cost})".format(self=self)class User(Base):__tablename__ = 'users'user_id = Column(Integer(), primary_key=True)username = Column(String(15), nullable=False, unique=True)email_address = Column(String(255), nullable=False)phone = Column(String(20), nullable=False)password = Column(String(25), nullable=False)created_on = Column(DateTime(), default=datetime.now)updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)def __init__(self, username, email_address, phone, password):self.username = usernameself.email_address = email_addressself.phone = phoneself.password = passworddef __repr__(self):return "User(username='{self.username}', "\"email_address='{self.email_address}', "\"phone='{self.phone}', "\"password='{self.password}')".format(self=self)class Order(Base):__tablename__ = 'orders'order_id = Column(Integer(), primary_key=True)user_id = Column(Integer(), ForeignKey('users.user_id'))shipped = Column(Boolean(), default=False)user = relationship('User', backref=backref('orders', order_by=order_id))def __repr__(self):return "Order(user_id={self.user_id}, " \"shipped={self.shipped})".format(self=self)class LineItems(Base):__tablename__ = 'line_items'line_items_id = Column(Integer(), primary_key=True)order_id = Column(Integer(), ForeignKey('orders.order_id'))cookie_id = Column(Integer(), ForeignKey('cookies.cookie_id'))quantity = Column(Integer())extended_cost = Column(Numeric(12,2))order = relationship('Order', backref=backref('line_items', order_by=line_items_id))cookie = relationship('Cookie', uselist=False)def __repr__(self):return "LineItems(order_id={self.order_id}, " \"cookie_id={self.cookie_id}, " \"quantity={self.quantity}, " \"extended_cost={self.extended_cost})".format(self=self)Base.metadata.create_all(engine)

插入新数据。c8trans_insert.py

from c8trans import session, User, Cookie, Order, LineItems
#两种cookie
cookiemon = User('cookiemon','mon@cookie.com','111-111-1111','password')
cc = Cookie('chocolate chip', 'http://some.aweso.me/cookie/recipe.html', 'cc01', 12, 0.50)
dcc = Cookie('dark chocolate chip', 'http://some.aweso.me/cookie/recipe_dark.html', 'cc02', 1, 0.75)
session.add(cc)
session.add(dcc)
session.commit()
#order1
o1 = Order()
o1.user = cookiemon
session.add(o1)
line1 = LineItems(order=o1, cookie=cc, quantity=9, extended_cost=1.50)
session.add(line1)
#order2
o2 = Order()
o2.user = cookiemon
session.add(o2)
line1 = LineItems(order=o2, cookie=cc, quantity=2, extended_cost=1.50)
line2 = LineItems(order=o2, cookie=dcc, quantity=9, extended_cost=6.75)
session.add(line1)
session.add(line2)session.commit()

ship_it函数

def ship_it(order_id):order = session.query(Order).get(order_id)for li in order.line_items:li.cookie.quantity = li.cookie.quantity - li.quantitysession.add(li.cookie)    #只有发生变化的对象才要addorder.shipped = Truesession.add(order)    #只有发生变化的对象才要addsession.commit()print('shipped order id:{}'.format(order_id))ship_it(1)    #正常
ship_it(2)    #报错，IntegrityError，quantity小于0违反约束了

异常会中断当前的会话，如果尝试通过这个会话发送更多语句（如查询）

try:ship_it(2)
except:print(session.query(Cookie.cookie_name, Cookie.quantity).all())

#报错
sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (sqlite3.IntegrityError) CHECK constraint failed: quantity_positive
[SQL: UPDATE cookies SET quantity=? WHERE cookies.cookie_id = ?]
[parameters: (-8, 4)]

继续使用中断的会话发送语句会触发InvalidRequestError。（当上一个事务没有正常结束时，会话就会卡住，影响后面的事务。）

处理异常：This Session's transaction has been rolled back due to a previous exception during flush.这句话表达不清楚，它的本意是需要我们手动回滚，而不是已经回滚了。To begin a new transaction with this Session, first issue Session.rollback()

注意回滚rollback的是事务transaction，而不是回滚会话。用rollback把这个问题事务清除，会话恢复正常。

try/except

from sqlalchemy.exc import IntegrityError
def ship_it(order_id):order = session.query(Order).get(order_id)for li in order.line_items:li.cookie.quantity = li.cookie.quantity - li.quantitysession.add(li.cookie)order.shipped = Truesession.add(order)try:session.commit()print('shipped order id:{}'.format(order_id))except IntegrityError as error:session.rollback()print('cookie is not enough')ship_it(2)
print(session.query(Cookie.cookie_name, Cookie.quantity).all())

sqlalchemy 8 会话与异常相关推荐

shell sqlplus执行sql文_如何通过 Shell 监控异常等待事件和活跃会话
作者 | JiekeXu 来源 | JiekeXu之路(ID: JiekeXu_IT) 转载请联系授权 | (微信ID:xxq1426321293) 大家好,我是 JiekeXu,分开这么久很高兴又和 ...
如何通过 Shell 监控异常等待事件和活跃会话
前几天有网友在墨天轮平台上问到"如何写一个定时任务监控用户会话连接数"的问题,由于当时比较忙,回答的比较简单也比较匆忙.最近也因为公司新项目老是加班,运维保障,安装 RAC.搭建 ...
使用SQLAlchemy创建数据模型
https://blog.csdn.net/happyanger6/article/details/53947162 如前所述,模型(models)是对数据抽象并提供通用访问接口的一种方式.在大多数网 ...
[转]SQLAlchemy Introduce
SQLAlchemy Introduce Tao Junjie 2015-12-13 19:17 Source perface 我们每天都要面对数据,数据库CRUD操作的能力对每个任务都至关重要.无论 ...
sqlalchemy相关操作
sqlalchemy是一个Python ORM工具,类似与Java中的Mybatis.Hibernate等ORM框架,这里主要介绍sqlalchemy的一些常用操作.这里使用的数据库为postgres ...
Sqlalchemy 使用总结
1.常用语句总结 # 查询语句 print(session.query(User).get({'id': 1})) # 根据id获取一条数据 print(session.query(User).fil ...
自学Python第十九天-flask框架
自学Python第十九天-flask框架安装和引用使用创建和运行应用设置应用处理函数及路由另一种路由注册唯一URL和重定向行为反向解析响应 get 和 post 请求 ,以及其他类型 ...
(用微信扫的静态链接二维码)微信native支付模式官方提供的demo文件中的几个bug修正...
原文:(用微信扫的静态链接二维码)微信native支付模式官方提供的demo文件中的几个bug修正 native支付模式一demo(用微信扫的静态链接二维码)BUG修复,一共4个BUG 1.nativ ...
ddos中的tcp反射攻击技术分析
我们常听说UDP反射攻击,那你听说过TCP反射攻击吗? 我们对TCP三次握手谙熟于心,但你确定服务器收到SYN包之后一定返回SYN/ACK吗? 现网的DDoS对抗中,基于TCP协议的反射攻击手法已经悄 ...
sniffer 和 debug flow
sniffer 和 debug flow sniffer 和 debug flow 复制模板,直接修改IP即可使用: diagnose sys session filter clear diagnos ...

sqlalchemy 8 会话与异常

sqlalchemy 8 会话与异常相关推荐

最新文章

热门文章

sqlalchemy 8 会话 与 异常

sqlalchemy 8 会话 与 异常相关推荐

最新文章

热门文章

sqlalchemy 8 会话与异常

sqlalchemy 8 会话与异常相关推荐