[sqlalchemy] IntegrityError on Association Proxy

[Erkan Özgür Yılmaz]

Hi,

I've encountered a situation where I don't know if this is a bug or it is
the correct behavior. It is about association_proxy. Please see the
User.__init__ and Department.__init__ methods.

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy


Base = declarative_base()

DBSession = scoped_session(
    sessionmaker(
        extension=None
    )
)


class Entity(Base):
    __tablename__ = 'Entities'

    id = Column(Integer, primary_key=True)
    entity_type = Column(String(128), nullable=False)

    __mapper_args__ = {
        "polymorphic_on": entity_type,
        "polymorphic_identity": "SimpleEntity"
    }


class User(Entity):
    __tablename__ = 'Users'
    __mapper_args__ = {"polymorphic_identity": "User"}
    user_id = Column('id', Integer, ForeignKey('Entities.id'),
                     primary_key=True)

    departments = association_proxy(
        'department_role',
        'department',
        creator=lambda d: DepartmentUsers(department=d)
    )

    department_role = relationship(
        'DepartmentUser',
        back_populates='user',
        cascade='all, delete-orphan',
        primaryjoin='Users.c.id==Department_Users.c.uid'
    )

    def __init__(self, departments=None):
        super(User, self).__init__()
        if departments is None:
            departments = []

        # using this creates an auto flush and raises an IntegrityError
        # self.departments = departments

        # where as using this is OK
        for department in departments:
            self.department_role.append(
                DepartmentUser(user=self, department=department)
            )


class Department(Entity):
    __tablename__ = 'Departments'
    __mapper_args__ = {"polymorphic_identity": "Department"}
    department_id = Column('id', Integer, ForeignKey('Entities.id'),
                           primary_key=True)

    users = association_proxy(
        'user_role',
        'user',
        creator=lambda u: DepartmentUser(user=u)
    )

    user_role = relationship(
        "DepartmentUser",
        back_populates="department",
        cascade='all, delete-orphan',
        primaryjoin='Departments.c.id==Department_Users.c.did'
    )

    @validates('user_role')
    def _validate_user_role(self, key, user_role):
        """validates the given user_role variable
        """
        return user_role

    def __init__(self, users=None):
        if users is None:
            users = []
        # again using this creates an auto flush and raises an IntegrityError
        # self.users = users

        # where as using this is OK
        for user in users:
            self.user_role.append(
                DepartmentUser(department=self, user=user)
            )


class Role(Entity):
    __tablename__ = 'Roles'
    __mapper_args__ = {"polymorphic_identity": "Role"}
    role_id = Column('id', Integer, ForeignKey('Entities.id'), primary_key=True)


class DepartmentUser(Base):
    __tablename__ = 'Department_Users'

    user_id = Column(
        'uid',
        Integer,
        ForeignKey('Users.id'),
        primary_key=True
    )

    user = relationship(
        'User',
        back_populates='department_role',
        primaryjoin='DepartmentUser.user_id==User.user_id'
    )

    department_id = Column(
        'did',
        Integer,
        ForeignKey('Departments.id'),
        primary_key=True
    )

    department = relationship(
        'Department',
        back_populates='user_role',
        primaryjoin='DepartmentUser.department_id==Department.department_id'
    )

    role_id = Column(
        'rid',
        Integer,
        ForeignKey('Roles.id'),
        nullable=True
    )

    role = relationship(
        'Role',
        primaryjoin='DepartmentUser.role_id==Role.role_id'
    )

    def __init__(self, department=None, user=None, role=None):
        self.department = department
        self.user = user
        self.role = role


if __name__ == "__main__":
    database_engine_settings = {
        "sqlalchemy.url": "sqlite:///:memory:",
        "sqlalchemy.echo": False,
    }

    engine = engine_from_config(database_engine_settings, 'sqlalchemy.')
    DBSession.configure(
        bind=engine,
        extension=None
    )
    Base.metadata.create_all(engine)

    user1 = User()
    dep1 = Department()
    DBSession.add(user1)
    DBSession.add(dep1)
    DBSession.commit()

    dep1.users.append(user1)

    assert user1 in dep1.users
    assert dep1 in user1.departments

    dep_users = DBSession.query(DepartmentUser).all()
    assert len(dep_users) == 1

    user2 = User()
    dep1.users.append(user2)
    DBSession.add(user2)
    DBSession.commit()

So, I get an IntegrityError if I set the **self.departments** or
**self.users** directly, so apparently creating the association objects
triggers an auto-flush but if I create the association objects instead of
directly setting the attributes there is no error. I'm ready to use the
work around of creating the association object but I'm also curious if this
is a bug or normal way of using association_proxy.

Thanks,

Erkan Ozgur Yilmaz

-- 
You received this message because you are subscribed to the Google Groups 
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.