简化模型的关键是使用associationproxy，因此您一定要检查一下。
假设 用户 在一个组中只能有一个 角色 ，则下面的代码应回答您的所有问题：
- 如何配置模型和关系
- 如何添加/删除/更新角色
- 如何检索您要求的数据（报告）
您应该接管模型部分，而忽略其余部分。下面是完全包含且有效的脚本：
from sqlalchemy import create_engine, Column, Integer, Unipre, ForeignKeyfrom sqlalchemy.orm import relationship, scoped_session, sessionmakerfrom sqlalchemy.orm.collections import attribute_mapped_collectionfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.ext.associationproxy import association_proxy# Configure test data SAengine = create_engine(u'sqlite:///:memory:', echo=False)session = scoped_session(sessionmaker(bind=engine, autoflush=False))base = declarative_base()class _baseMixin(object): """ A helper mixin class to set properties on object creation. Also provides a convenient default __repr__() function, but be aware that also relationships are printed, which might result in loading the relation objects from the database """ def __init__(self, **kwargs): for k, v in kwargs.items(): setattr(self, k, v) def __repr__(self): return "<%s(%s)>" % (self.__class__.__name__, ', '.join('%s=%r' % (k, self.__dict__[k])for k in sorted(self.__dict__)if '_' != k[0]#if '_sa_' != k[:4] and '_backref_' != k[:9]) )# relation creator factory functionsdef _creator_gr(group, role): res = UserGroup(group=group, role=role) return resdef _creator_ur(user, role): res = UserGroup(user=user, role=role) return res############################################################################### Object Model##############################################################################class Role(base, _baseMixin): __tablename__ = 'roles' # columns id = Column(Integer, primary_key=True, autoincrement=True) name = Column(Unipre(16), unique=True) # relations usergroup = relationship("UserGroup", backref='role')class User(base, _baseMixin): __tablename__ = 'users' # columns id = Column(Integer, primary_key=True, autoincrement=True) name = Column(Unipre(16), unique=True) # relations _rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('group'), cascade='all,delete-orphan', backref='user', ) groups = association_proxy('_rel_usergroup', 'role', creator=_creator_gr)class Group(base, _baseMixin): __tablename__ = 'groups' # columns id = Column(Integer, primary_key=True, autoincrement=True) name = Column(Unipre(16), unique=True) # relations _rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('user'), cascade='all,delete-orphan', backref='group', ) users = association_proxy('_rel_usergroup', 'role', creator=_creator_ur)class UserGroup(base, _baseMixin): __tablename__ = 'user_group' # columns id = Column(Integer, primary_key=True, autoincrement=True) user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False) group_id = Column(Integer, ForeignKey('groups.id', ondelete='CASCADE'), nullable=False) role_id = Column(Integer, ForeignKey('roles.id', ondelete='CASCADE'), nullable=False) # relations # (all backrefs)############################################################################### TESTS (showing usages)## Requirements:# - list all groups of the user: user.groups (use keys)# - list all users of the group: group.users (use keys)# - get all users ordered (grouped) by group with the role title##############################################################################def _requirement_get_user_groups(user): return user.groups.keys()def _requirement_get_group_users(group): return group.users.keys()def _requirement_get_all_users_by_group_with_role(): qry = session.query(Group).order_by(Group.name) res = [] for g in qry.all(): for u, r in sorted(g.users.items()): value = (g.name, u.name, r.name) res.append(value) return resdef _test_all_requirements(): print '--requirement: all-ordered:' for v in _requirement_get_all_users_by_group_with_role(): print v print '--requirement: user-groups:' for v in session.query(User).order_by(User.id): print v, " has groups: ", _requirement_get_user_groups(v) print '--requirement: group-users:' for v in session.query(Group).order_by(Group.id): print v, " has users: ", _requirement_get_group_users(v)# create db schemabase.metadata.create_all(engine)############################################################################### CREATE TEST DATA############################################################################### create entitiesu_peter = User(name='u_Peter')u_sonja = User(name='u_Sonja')g_sales = Group(name='g_Sales')g_wales = Group(name='g_Wales')r_super = Role(name='r_Super')r_minor = Role(name='r_Minor')# helper functionsdef _get_entity(entity, name): return session.query(entity).filter_by(name=name).one()def get_user(name): return _get_entity(User, name)def get_group(name): return _get_entity(Group, name)def _checkpoint(): session.commit() session.expunge_all() _test_all_requirements() session.expunge_all() print '-' * 80# test: **ADD**u_peter.groups[g_wales] = r_minor # addg_wales.users[u_sonja] = r_super # addg_sales.users[u_peter] = r_minor # addsession.add(g_wales)#session.add(g_sales)_checkpoint()# test: **UPDATE**u_peter = get_user('u_Peter')assert u_peter.name == 'u_Peter' and len(u_peter.groups) == 2assert len(u_peter.groups) == 2g_wales = get_group('g_Wales')g_wales.users[u_peter] = r_super # update_checkpoint()# test: **DELETE**u_peter = get_user('u_Peter')assert u_peter.name == 'u_Peter' and len(u_peter.groups) == 2g_wales = get_group('g_Wales')del u_peter.groups[g_wales] # delete_checkpoint()
