passing additional values to association_proxy's 'creator' method

[Gerald Thibault]

I have a User model, and a Group model, and a m2m relation between them.

I also have a proxy on the user that routes directly to the Group instances. The creator function, however, only takes one param (group), so when I pass in a group, I have no way to look it up to see if the (user_id, group_id) combo is in use already, because i only have the group. It seems like I am probably missing something very simple, but I can't figure out what it is. The code below can be run to show the issue I am having.

The last line, user.groups = [group] , throws an error because 'self' isn't defined. self clearly doesn't work here, what would the workaround be to get the User instance's "id" value into that creator function? The group I am passing in is the same as the group currently assigned, so rather than purging the collection and adding a UserGroup with only a group_id (the user_id gets populated later via the relationship, but at that point it is too late, because it's already decided the new object is not the same as the old, and it purges the old one), i'd like to be able to identify that the identity_key is the same, and just let it be.

####Code below######

from sqlalchemy import *

from sqlalchemy.ext.associationproxy import association_proxy

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import Session, relationship, backref

e = create_engine('sqlite:////tmp/test.db', echo=False)

Base = declarative_base()

Base.metadata = MetaData(e)

class User(Base):

    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)

    def usergroup_creator(group):

        session = Session(e)

        ugroup = session.query(UserGroup) \

            .filter_by(user_id=self.id) \

            .filter_by(group_id=group.id) \

            .first()

        if not ugroup:

            return UserGroup(group=group)

    groups = association_proxy('user_groups', 'group',

        creator=usergroup_creator)

class Group(Base):

    __tablename__ = 'groups'

    id = Column(Integer, primary_key=True)

    

class UserGroup(Base):

    __tablename__ = 'user_groups'

    user_id = Column(Integer, ForeignKey(User.id), primary_key=True)

    group_id = Column(Integer, ForeignKey(Group.id), primary_key=True)

    user = relationship(User, backref='user_groups')

    group = relationship(Group, backref='user_groups')

    

if __name__ == '__main__':

    Base.metadata.drop_all()

    Base.metadata.create_all()

    

    user = User(id=1)

    group = Group(id=1)

    ugroup = UserGroup(user=user, group=group)

    session = Session(e)

    session.add(user)

    session.add(group)

    session.add(ugroup)

    session.commit()

    session.expunge_all()

    

    user = session.query(User).get(1)

    group = session.query(Group).get(1)

    user.groups = [group]

[Michael Bayer]

On May 16, 2013, at 8:59 PM, Gerald Thibault <diesel...@gmail.com> wrote:

> user = User(id=1)
> group = Group(id=1)
> ugroup = UserGroup(user=user, group=group)
>
> session = Session(e)
> session.add(user)
> session.add(group)
> session.add(ugroup)
> session.commit()
> session.expunge_all()
>
> user = session.query(User).get(1)
> group = session.query(Group).get(1)
> user.groups = [group]

So the reassignment of the same "group" under typical conditions works, because the usual pattern is to set delete-orphan cascade on the UserGroup:

class UserGroup(Base):
# ...

user = relationship(User, backref=backref('user_groups', cascade="all, delete-orphan"))
group = relationship(Group, backref=backref('user_groups', cascade="all, delete-orphan"))

class User(Base):
# ...

groups = association_proxy('user_groups', 'group', creator=lambda group: UserGroup(group=group))

what will happen there is, when you say user.groups = [group], it will basically remove the previous UserGroup, delete it due to the delete-orphan, then create a new one.

Now, lets say you don't want that delete. The next pattern is, use a set() instead of a list and use update():

class UserGroup(Base):
# ...

user = relationship(User, backref=backref('user_groups',
collection_class=set))
group = relationship(Group)

# to add groups:

user.groups.update([group])

that's the cleanest way to do it and doesn't even need a SQL statement.

But OK lets say you don't like that you can't use lists and can't use assignment and other things are going on, then we need to do your lookup in a validation event, which is pretty much what you asked for in the first place. Here are the special things to consider:

1. We need to use the UserGroup as loaded from the current session (object_session(self)), and if there is no current session we assume there's no dupes (though you're free to search around in the list here locally if the User is transient/pending still).

2. We need to prevent the flush from happening here else our to-be-rejected UserGroup gets flushed.

3. We need to evict that to-be-rejected UserGroup also, again else it gets flushed (or we can use delete-orphan cascade which would have the same effect):

from sqlalchemy.orm import validates, object_session

class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)

groups = association_proxy('user_groups', 'group',

creator=lambda group: UserGroup(group=group))

@validates("user_groups")
def _validate_user_group(self, key, value):
session = object_session(self)
if session is not None:
group = value.group
with session.no_autoflush:

ugroup = session.query(UserGroup) \
.filter_by(user_id=self.id) \
.filter_by(group_id=group.id) \
.first()

if ugroup:
session.expunge(value)
return ugroup

return value

class UserGroup(Base):
__tablename__ = 'user_groups'
user_id = Column(Integer, ForeignKey(User.id), primary_key=True)
group_id = Column(Integer, ForeignKey(Group.id), primary_key=True)

user = relationship(User, backref='user_groups')
group = relationship(Group, backref='user_groups')

# .. later...