SQLAlchemy необычный не нулевой сбой при добавлении в отношения

Question

У меня очень странная ошибка с SQLAlchemy.Я удалил как можно больше своего кода, чтобы сузить проблему, а дальнейшее удаление приведет к исчезновению ошибки.Я смог воспроизвести проблему на другом ПК с новой установкой SQLAlchemy (на Python 2.7).

Если я добавлю что-то вроде этого:

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()

Iполучить эту ошибку (если я не удаляю что-либо еще из кода):

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()

Я буквально не знаю, что происходит.Как заявление об отказе от ответственности, другой протестированный мной компьютер находится в той же рабочей сети, но, поскольку я тестирую с 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
• Удалить if 'DATABASE_URL' not in os.environ:
• Заменить create_engine(os.environ['DATABASE_URL']) на create_engine('sqlite://')
• Удалить Category
• Удалить отношения из Category
• Удалить fullname, __init__ или __repr__ из Category
• Удалить __init__, __repr__, __eq__ или __neq__ из ApplicationVersion

Буду признателен за любую помощь, так как она сводит меня с ума.Я могу обойти эту проблему, используя 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)

В частности, вы допускаете значение по умолчанию None для ApplicationVersion.application. Я не уверен, какое значение это служит вам, поскольку конструктор по умолчанию не требует, чтобы вы передавали явное значение для любого из полей модели, и поэтому, если оно не предоставлено, оно будет None при доступе в любом случае.

Тогда в этой строке в вашем тесте:

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

... вы явно создаете объект ApplicationVersion с application=None из-за конструктора, но в то же время добавляете его в коллекцию python.versions. Способ, которым эти отношения разрешают значения внешнего ключа, кажется непоследовательным, поэтому иногда он пытается очистить с помощью application_id=1, который является pk нового объекта Application, а в других случаях он пытается очистить с 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__(), и он будет работать постоянно.