用户,组和角色的多对多声明式SQLAlchemy定义 [英] Many-to-many declarative SQLAlchemy definition for users, groups, and roles
问题描述
我是SQLAlchemy的新手,我想知道定义此类表和关系的最佳方法是什么.我希望能够通过user.groups
访问来自用户的组,通过group.users
访问组中的用户,并找出用户在组中的角色(我假设这将是在关联模型中定义的逻辑) ).我还希望按组分组选择所有用户,并包括角色标题.
I'm new to SQLAlchemy, and was wondering what the best way to go about defining such tables and a relationship would be. I want to be able to access groups from a user by user.groups
, users in a group by group.users
, and find out the role of a user in a group (which I'm assuming will be logic defined in an association model). I also want to select all users, group by group, and include role titles.
我尝试使用本教程创建关联表(声明式),并删除了User
和Group
类中relationship
属性的secondary
参数,但是那样我将失去直接从用户访问组,以及直接从组访问用户(关联表中的backref
指向关联类本身:/).
I've tried using the tutorial to create the association table (declarative style) and removing the secondary
argument to relationship
properties in the User
and Group
classes, but then I would lose the ability to access groups directly from users, and users directly from groups (the backref
's in the association table point to the association class itself :/).
任何帮助将不胜感激.
推荐答案
The key to simplifying your model is to use associationproxy, therefore you should definitely check it out.
Assuming that the user can have only one role within a group, the code below should answer all your questions:
- 如何配置模型和关系
- 如何添加/删除/更新角色
- 如何检索您要求的数据(报告)
您应该接管模型部分,而忽略其余部分.下面包含完整且有效的脚本:
You should take over the model part and forget the rest. The completely contained and working script below:
from sqlalchemy import create_engine, Column, Integer, Unicode, ForeignKey
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy
# Configure test data SA
engine = 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 functions
def _creator_gr(group, role):
res = UserGroup(group=group, role=role)
return res
def _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(Unicode(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(Unicode(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(Unicode(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 res
def _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 schema
Base.metadata.create_all(engine)
##############################################################################
# CREATE TEST DATA
##############################################################################
# create entities
u_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 functions
def _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 # add
g_wales.users[u_sonja] = r_super # add
g_sales.users[u_peter] = r_minor # add
session.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) == 2
assert len(u_peter.groups) == 2
g_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) == 2
g_wales = get_group('g_Wales')
del u_peter.groups[g_wales] # delete
_checkpoint()
这篇关于用户,组和角色的多对多声明式SQLAlchemy定义的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!