python - 附加到关系时 SQLAlchemy 异常非空失败

Question

我在使用 SQLAlchemy 时遇到了一个非常奇怪的错误。我已经删除了尽可能多的代码以缩小问题范围，删除更多代码将导致错误消失。我能够通过全新的 pip 安装 SQLAlchemy(在 Python 2.7 上)在另一台 PC 上重现该问题。

如果我做任何附加这样的事情的变体:

python = Application(name='Python')
python.versions.append(ApplicationVersion(version=27))
session.add(python)
session.commit()

#or

python = Application(name='Python')
session.add(python)
session.commit()
python.versions.append(ApplicationVersion(version=27))
session.commit()

我收到此错误(如果我不从代码中删除任何其他内容):

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: ApplicationVersion.application_id
[SQL: INSERT INTO "ApplicationVersion" (application_id, version_int) VALUES (?, ?)]
[parameters: (None, 27)]

但是，正如我提到的，如果我删除任何内容，它会完美运行。例如，通过删除下面这个函数的文档字符串，它将正确分配 application_id 并按预期工作。

@contextmanager
def Session():
    """Setup session to allow for usage with a context manager."""
    session = _Session()
    yield session
    session.close()

我完全不知道发生了什么。作为免责声明，我测试的另一台 PC 在同一个工作网络上，但由于我正在使用 sqlite 进行测试，我无法想象它是基于网络的东西。

这是重现错误的代码(它是一个由多个文件连接而成的文件):

######### CONNECT.PY #######
import os
from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import sessionmaker


class BaseTable(object):
    """General things to apply to each table.

    Help: https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/mixins.html
    """
    @declared_attr
    def __tablename__(cls):
        """Set the table name to that of the model."""
        return cls.__name__


if 'DATABASE_URL' not in os.environ:
    os.environ['DATABASE_URL'] = 'sqlite://'

Engine = create_engine(os.environ['DATABASE_URL'])

Base = declarative_base(bind=Engine, cls=BaseTable)

_Session = sessionmaker(bind=Base.metadata.bind)


@contextmanager
def Session():
    """Setup session to allow for usage with a context manager."""
    session = _Session()
    yield session
    session.close()


########## MODELS.PY ###########
import time
import os
from sqlalchemy import Column, Integer, SmallInteger, String, Text
from sqlalchemy import ForeignKey, UniqueConstraint, Table, event
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import backref, relationship, validates


class Category(Base):
    row_id = Column(Integer, primary_key=True)
    name = Column(String(64), nullable=False)
    parent_id = Column(Integer, ForeignKey('Category.row_id'), nullable=True)

    parent = relationship('Category', foreign_keys=parent_id, remote_side=row_id)
    children = relationship('Category')

    __table_args__ = (
        UniqueConstraint('name', 'parent_id', name='unique_name_parent'),
    )

    @hybrid_property
    def fullname(self):
        parent = self.parent
        visited = set()
        chain = [self.name]
        while parent:
            if parent in visited:
                break
            visited.add(parent)
            chain.append(parent.name)
            parent = parent.parent

        return '.'.join(chain[::-1])

    def __init__(self, name, collection, parent=None, creator=None, **kwargs):
        super(Category, self).__init__(name=name, collection=collection, parent=parent, creator=creator, **kwargs)

    def __repr__(self):
        return '<{cls} "{fullname}">'.format(
            cls=self.__class__.__name__,
            fullname=self.fullname,
        )


class Application(Base):
    row_id = Column(Integer, primary_key=True)
    name = Column(String(16), nullable=False)
    versions = relationship('ApplicationVersion', order_by='ApplicationVersion.version_int')


class ApplicationVersion(Base):
    row_id = Column(Integer, primary_key=True)
    application_id = Column(Integer, ForeignKey('Application.row_id'), nullable=False)
    version_int = Column(Integer, nullable=False)

    application = relationship('Application', foreign_keys=application_id)

    __table_args__ = (
        UniqueConstraint('application_id', 'version_int', name='unique_application_version'),
    )

    def __init__(self, version, application=None, **kwargs):
        super(ApplicationVersion, self).__init__(application=application, version_int=version, **kwargs)

    def __repr__(self):
        return '<{cls} "{application} {version}">'.format(
            cls=self.__class__.__name__,
            application=self.application.name,
            version=self.version_int,
        )

    def __eq__(self, num):
        return self.version_int == num

    def __neq__(self, num):
        return self.version_int != num


######## TEST.PY ########
Base.metadata.create_all()

if __name__ == '__main__':
    with Session() as session:

        # Setup programs and versions
        python = Application(name='Python')
        python.versions.append(ApplicationVersion(version=27))
        session.add(python)
        session.commit()

        print python.versions

这些是将停止错误的各种操作:

• 从 BaseTable 或 Session 中删除文档字符串
• 删除 如果 'DATABASE_URL' 不在 os.environ 中:
• 将 create_engine(os.environ['DATABASE_URL']) 替换为create_engine('sqlite://')
• 删除类别
• 从类别中删除关系
• 从类别中删除fullname、__init__ 或__repr__
• 从中删除 __init__、__repr__、__eq__ 或 __neq__应用程序版本

任何帮助将不胜感激，因为它让我有点发疯。我可以通过使用 session.add(ApplicationVersion(python, 27)) 来解决这个问题，但我想知道这里到底发生了什么，因为我从未见过 Python 的行为是这样的之前。

Answer 1

我发现问题出在您在 ApplicationVersion 上定义的自定义构造函数:

def __init__(self, version, application=None, **kwargs):
    super(ApplicationVersion, self).__init__(application=application, version_int=version, **kwargs)

具体来说，您允许 ApplicationVersion.application 的默认值为 None。我不确定这对您有什么值(value)，因为默认构造函数不要求您为模型的任何字段传递显式值，因此如果未提供，它将是 None 当无论如何访问。

然后在测试的这一行中:

python.versions.append(ApplicationVersion(version=27))

... 由于构造函数，您使用 application=None 显式创建了一个 ApplicationVersion 对象，但同时，将其附加到 python.versions 集合。这些关系解析外键值的方式似乎不一致，因此有时它会尝试使用 application_id=1 进行刷新，这是新的 Application 对象的 pk，并且其他时候，它会按照构造函数的指示尝试使用 application_id=None 刷新。但是 application_id 不可为空:

application_id = Column(Integer, ForeignKey('Application.row_id'), nullable=False)

...这就是您收到 IntegrityError 的时候。

SQLAlchemy 必须在将关系属性显式设置为 None 和它根本没有设置之间做出一些区分，因为如果您停止将 application 设置为 None 在你的构造函数中，问题停止了:

def __init__(self, version, **kwargs):
    super(ApplicationVersion, self).__init__(version_int=version, **kwargs)

我能够将您的示例缩减为这个通用示例(抱歉，Python 3，因此您需要调整 print 调用):

from sqlalchemy import create_engine, Column, Integer
from sqlalchemy import ForeignKey
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

engine = create_engine('sqlite://')

Base = declarative_base()

Session = sessionmaker(bind=engine)


class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship('Child')


class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    num = Column(Integer)
    parent_id = Column(Integer, ForeignKey('parent.id'), nullable=False)
    parent = relationship('Parent')

    def __init__(self, parent=None, **kwargs):
        super(Child, self).__init__(parent=parent, **kwargs)


if __name__ == '__main__':
    Base.metadata.create_all(engine)
    error_cnt = 0
    success_cnt = 0
    for _ in range(20):
        s = Session()
        try:
            parent = Parent()
            parent.children.append(Child())
            s.add(parent)
            s.commit()
        except IntegrityError:
            error_cnt += 1
        else:
            success_cnt += 1
        finally:
            s.close()
    print('errors', error_cnt)
    print('successes', success_cnt)

当你运行它时，你应该得到成功和错误的随机计数。然后删除 Child.__init__() 方法，它一直有效。