SQAlchemy Native Secondary Relationship with Composite Primary Keys

Question

There are 3 table: Account, Role, User. Both Roleand Userhave a foreign key account_idthat points to Account.

A user can have several roles, therefore a table roles_usersthat acts as a table of secondary relations between Roleand User.

The Accounttable is the tenant table for our application, it is used to separate different customers.

Note that all tables have (except Account) composite primary keys with account_id. This is done for several reasons, but let me say that this is done so that everything is agreed.

Now, if I have a simple secondary connection ( User.roles- the one that is commented out), everything works as expected. Well, sort of ... it raises a legitimate warning (although I believe that this should be a mistake):

SAWarning: relationship 'User.roles' will copy column role.account_id to column roles_users.account_id, which conflicts with relationship(s): 'User.roles' (copies user.account_id to roles_users.account_id). Consider applying viewonly=True to read-only relationships, or provide a primaryjoin condition marking writable columns with the foreign() annotation.

That's why I created a second relationship User.roles- one that is not commented out. The request works as expected, which has 2 conditions for joining and that's it. However, I get this error when I try to save some roles for the user:

sqlalchemy.orm.exc.UnmappedColumnError: Can't execute sync rule for source column 'roles_users.role_id'; mapper 'Mapper|User|user' does not map this column.  Try using an explicit `foreign_keys` collection which does not include destination column 'role.id' (or use a viewonly=True relation).

As far as I understand, SA cannot figure out how to save the secondary because it has custom primaryjoinand secondaryjointherefore it suggests using viewonly=True, which leads to a simple ignoring of the role relationship when saving the model.

, ( ). , . .

User.roles = some_roles, primaryjoin secondaryjoin ?

SA 1.1.9:

from sqlalchemy import create_engine, Column, Integer, Text, Table, ForeignKeyConstraint, ForeignKey, and_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import foreign, relationship, Session

Base = declarative_base()


class Account(Base):
    __tablename__ = 'account'
    id = Column(Integer, primary_key=True)


roles_users = Table(
    'roles_users', Base.metadata,
    Column('account_id', Integer, primary_key=True),
    Column('user_id', Integer, primary_key=True),
    Column('role_id', Integer, primary_key=True),

    ForeignKeyConstraint(['user_id', 'account_id'], ['user.id', 'user.account_id']),
    ForeignKeyConstraint(['role_id', 'account_id'], ['role.id', 'role.account_id']),
)


class Role(Base):
    __tablename__ = 'role'
    id = Column(Integer, primary_key=True)
    account_id = Column(Integer, ForeignKey('account.id'), primary_key=True)
    name = Column(Text)

    def __str__(self):
        return '<Role {} {}>'.format(self.id, self.name)


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    account_id = Column(Integer, ForeignKey('account.id'), primary_key=True)
    name = Column(Text)

    # This works as expected: It saves data in roles_users
    # roles = relationship(Role, secondary=roles_users)

    # This custom relationship - does not work
    roles = relationship(
        Role,
        secondary=roles_users,
        primaryjoin=and_(foreign(Role.id) == roles_users.c.role_id,
                         Role.account_id == roles_users.c.account_id),
        secondaryjoin=and_(foreign(id) == roles_users.c.user_id,
                           account_id == roles_users.c.account_id))


engine = create_engine('sqlite:///')
Base.metadata.create_all(engine)
session = Session(engine)

# Create our account
a = Account()
session.add(a)
session.commit()

# Create 2 roles
u_role = Role()
u_role.id = 1
u_role.account_id = a.id
u_role.name = 'user'
session.add(u_role)

m_role = Role()
m_role.id = 2
m_role.account_id = a.id
m_role.name = 'member'
session.add(m_role)
session.commit()

# Create 1 user
u = User()
u.id = 1
u.account_id = a.id
u.name = 'user'

# This does not work
u.roles = [u_role, m_role]
session.add(u)
session.commit()

# Works as expected
i = roles_users.insert()
i = i.values([
    dict(account_id=a.id, role_id=u_role.id, user_id=u.id),
    dict(account_id=a.id, role_id=m_role.id, user_id=u.id),
])
session.execute(i)

# re-fetch user from db
u = session.query(User).first()
for r in u.roles:
    print(r)

. primaryjoin secondaryjoin .