Hi,
The situation is that I have tow tables (User and UserGrp) and two
association tables (one for many-to-many between User and UserGrp, one for
many-to-many between UserGrp and user group permissions). Each user can
belong to multiple user groups, and each user group can have multiple
permissions. The permissions that a user has is the union of the
permissions of the groups that it belongs to.
I cached the permission of each user in the User table (see the *
User._perms_cache* attribute). So it is necessary to invalidate the cache
when the relationship between users and user groups gets changed. To
invalidate the cache, I do an UPDATE on all the affected users, and I also
have to expire the _perms_cache attribute of all persistent User instances.
However, I searched the docs and couldn't find an appropriate API (*
Session.expire_all* does not work, which is shown later;
*Session.expire*requires an instance).
I looked into SQLAlchemy's source, and finally found that I can iterate
over all the persistent objects via *Session.identity_map*. But this is not
documented, and do I really have to do this?
The simplified model is attached (sorry...it's still that long....)
And I also wonder why the following does not work? It raises
an IntegrityError.
python test.py
>>> g0.perms.add(5)
>>> ses.expire_all()
>>> ses.commit()
Finally, thanks very much for your patient reading!
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To view this discussion on the web visit
https://groups.google.com/d/msg/sqlalchemy/-/tJCdWyGW2nQJ.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
from sqlalchemy import Column, event
from sqlalchemy.types import Integer, Text
from sqlalchemy.schema import ForeignKey
from sqlalchemy.orm import relationship, backref
from sqlalchemy.orm.session import object_session
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MapUserGrpAndGrpPerm(Base):
__tablename__ = 'map_ugrp_grpperm'
def __init__(self, perm):
self.perm = perm
gid = Column(Integer, ForeignKey('ugrp.id'), primary_key = True)
perm = Column(Integer, primary_key = True)
class MapUserAndUserGrp(Base):
__tablename__ = 'map_user_ugrp'
uid = Column(Integer, ForeignKey('user.id'), primary_key = True)
gid = Column(Integer, ForeignKey('ugrp.id'), primary_key = True)
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key = True)
groups = relationship('UserGrp', secondary = 'map_user_ugrp')
_perms_cache_rst = None
_perms_cache = Column('permscache', Text())
@property
def perms(self):
if self._perms_cache_rst is None:
self._perms_cache_rst = self._get_perms()
return self._perms_cache_rst
def _get_perms(self):
if self._perms_cache is None:
rst = set()
for i in self.groups:
rst.update(i.perms)
rst = frozenset(rst)
self._perms_cache = '|' . join([str(i) for i in rst])
return rst
return frozenset([int(i) for i in self._perms_cache.split('|')])
class UserGrp(Base):
__tablename__ = 'ugrp'
id = Column(Integer, primary_key = True)
_perms = relationship('MapUserGrpAndGrpPerm', collection_class = set)
perms = association_proxy('_perms', 'perm')
def invalidate_user_perm_cache(session, gid):
for (cls, pk), obj in session.identity_map.iteritems():
if cls is User:
session.expire(obj, ['_perms_cache'])
obj._perms_cache_rst = None
sub = session.query(MapUserAndUserGrp.uid) \
.filter(MapUserAndUserGrp.gid == gid)
session.query(User).filter(User.id.in_(sub)) \
.update({User._perms_cache: None}, synchronize_session = False)
def _invcache_on_grp_perm_chg(target, *args):
ses = object_session(target)
if ses is not None:
invalidate_user_perm_cache(ses, target.id)
#for i in 'append', 'remove', 'set':
# event.listen(UserGrp._perms, i, _invcache_on_grp_perm_chg)
if __name__ == '__main__':
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:', echo = True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind = engine)
ses = Session()
g0 = UserGrp()
g1 = UserGrp()
u0 = User()
u0.groups.append(g0)
u0.groups.append(g1)
g0.perms.update([1, 2])
g1.perms.update([2, 3])
ses.add(u0)
ses.commit()
import code
code.interact(local = locals())
