SQLAlchemy 模型中数据的错误表示

在这里插入图片描述

1. 问题背景

在使用 SQLAlchemy 0.6.0 版本（也曾尝试使用 0.6.4 版本）的 Pylons 应用程序中遇到了一个 SQLAlchemy ORM 问题。
该问题出现在使用 psycopg2 作为数据库驱动程序、连接至 Postgresql 8.2 数据库的环境中。
定义了一个 User 模型对象，其中包含以下属性：

class User(Base):__tablename__ = 'users'__table_args__ = (saschema.UniqueConstraint("login", "company_id"), {})__mapper_args__ = {'order_by' : 'lower(users.name)',}user_id = Column(Integer, primary_key=True)email = Column(String)login = Column(String)company_id = Column(String, ForeignKey('company.company_id'))

尝试使用以下代码更新 User 模型的实例：

def do_update(user_id):existing = Session().query(User).filter_by(user_id=user_id).one()for field in ('login', 'email', 'name', 'is_admin_user'): # only email changes; it's set to "" from "foo@bar.com"if field in params:setattr(existing, field, params[field])if 'advanis_portal_user_id' in params:if not existing.portal_link:existing.portal_link = UserPortalLink()existing.portal_link.advanis_portal_user_id = params['advanis_portal_user_id']if 'password' in params and existing.password:existing.password.password = Password.encrypt(existing.login, params['password'])UserValidator(existing) # raises an exceptionself._commit()return existing

当电子邮件地址从 “foo@bar.com” 变更为 “” 时，UserValidator 会引发异常，随后，即使 Pylons 服务器重启，通过以下查询返回的用户电子邮件地址仍为空白：

Session().query(User).filter_by(login=login, company_id=company).one()
Session().query(User).all()

通过以下查询可以返回电子邮件地址完整的用户：

Session().query(User).filter_by(user_id=user_id).one()

2. 解决方案

问题的原因是当电子邮件字段被设置为 “” 时，SQLAlchemy ORM 不会将该更改持久化到数据库中。这可能是由于在设置电子邮件字段为空字符串之前没有调用 session.flush() 方法造成的。调用 session.flush() 方法可以将未提交的更改写入到数据库中，从而确保当对数据库发出查询时可以获取到最新的数据。

为了解决这个问题，需要在代码中调用 session.flush() 方法，如下所示：

def do_update(user_id):existing = Session().query(User).filter_by(user_id=user_id).one()for field in ('login', 'email', 'name', 'is_admin_user'): # only email changes; it's set to "" from "foo@bar.com"if field in params:setattr(existing, field, params[field])if 'advanis_portal_user_id' in params:if not existing.portal_link:existing.portal_link = UserPortalLink()existing.portal_link.advanis_portal_user_id = params['advanis_portal_user_id']if 'password' in params and existing.password:existing.password.password = Password.encrypt(existing.login, params['password'])UserValidator(existing) # raises an exception# Add this line to flush the changes to the databasesession.flush()self._commit()return existing