On May 16, 2013, at 8:59 PM, Gerald Thibault <dieselmach...@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...
user = session.query(User).get(1)
group = session.query(Group).get(1)
user.groups = [group]
--
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?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.
