SQLAlchemy 模型中数据的错误表示

news2024/12/27 0:57:55

在这里插入图片描述

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 exception
    self._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 database
    session.flush()

    self._commit()
    return existing