|
我是SQLAlchemy的新手,我想知道定义此类表和关系的最佳方法是什么。我希望能够从,,,组中的用户访问user.groups用户的组group.users,并找出用户在组中的角色(我假设这将是在关联模型中定义的逻辑)。我还希望按组分组选择所有用户,并包括角色标题。
! i7 X8 E, {/ E) S1 A, `我已经使用教程创建关联表(声明样式),并删除试图secondary参数relationship的属性User和Group类别,但随后我就失去了能力,直接从用户的访问组,并直接从团体用户(backref的在关联表中指向关联类本身:/)。
! }/ E# f* \4 u8 t( Q( |4 o: o任何帮助,将不胜感激。# K8 U0 b# \: x4 D; f: j* l
2 P6 m$ [! m1 y4 o6 l
解决方案:
a9 b* e; n$ ?" y* k& V
5 a2 ?7 d$ e* f2 n
7 |9 c; f% p: m7 v' U9 x& _- r) m2 F3 j2 |. Z
简化模型的关键是使用associationproxy,因此您一定要检查一下。
0 C$ J% n, g/ F' I: j3 Q假设 用户 在一个组中只能有一个 角色 ,则下面的代码应回答您的所有问题:
+ e5 P( c1 s" Z8 v! v如何配置模型和关系4 [; n& ? V2 ~ a" v8 P3 {
如何添加/删除/更新角色# E% P, M0 u) z3 |" N4 x' i$ r) w
如何检索您要求的数据(报告)
) I% H+ R: c) W% {0 V" z* ]0 ]0 V6 c# y" @ w, C
您应该接管模型部分,而忽略其余部分。下面是完全包含且有效的脚本:
% O# b3 Y% P8 A f2 R' _/ C- Dfrom sqlalchemy import create_engine, Column, Integer, Unicode, ForeignKey& [! ]! | d' b$ k$ W
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
; L3 C- N5 X; \" B9 ifrom sqlalchemy.orm.collections import attribute_mapped_collection
0 d% \. ~+ |$ w/ u7 y- |0 K2 xfrom sqlalchemy.ext.declarative import declarative_base% ?9 m: V* u u1 S- N W
from sqlalchemy.ext.associationproxy import association_proxy
/ ?# |; d: X& R* M1 G* z: W# Configure test data SA1 g" ~* J! i/ z' _. R
engine = create_engine(u'sqlite:///:memory:', echo=False)
& p! F* @/ A6 t, `; v" r' Dsession = scoped_session(sessionmaker(bind=engine, autoflush=False))0 e& u. D, s" u0 e x' w
Base = declarative_base()
2 J/ }2 d8 C! X" R ^) B' H) M; dclass _BaseMixin(object):
: f( m: A: e; F8 b: x8 ?. n5 O """
: E6 i. n6 Q* d8 y" M$ X A helper mixin class to set properties on object creation.
+ e* O7 h$ V) i: I- `0 r8 n Also provides a convenient default __repr__() function, but be aware that& E0 g- \- D" h' q. g
also relationships are printed, which might result in loading the relation
6 S% }* Y3 F D ] ? objects from the database
. N, J4 S6 t; v: c* [ M """- z4 R& f" |# P" {4 K
def __init__(self, **kwargs):( J6 X7 X* ~ u0 y
for k, v in kwargs.items():
2 t) {+ N* C n9 O/ `+ b. v, x) o! H setattr(self, k, v) K& k' X; s+ J9 [7 t
def __repr__(self):
( c! t# M& {* j0 g; D/ M return "" % (self.__class__.__name__,
7 e: c9 \8 c5 P ', '.join('%s=%r' % (k, self.__dict__[k])1 q9 Q1 v1 _; w+ n8 v7 P0 G ~
for k in sorted(self.__dict__)
) d/ h* E" |* `% i if '_' != k[0]
" T$ `- D' A! \1 ]) J# _$ | #if '_sa_' != k[:4] and '_backref_' != k[:9]
) f+ F7 c. g" W) ~9 q )
& U7 Q0 M3 P- G k# K )
! z9 N1 f+ k8 X7 b' P7 r( z+ _! c6 p9 U# k6 e" Q7 W4 E( E
# relation creator factory functions8 M7 G7 `+ h% r
def _creator_gr(group, role):2 N$ q' \; h2 b4 V
res = UserGroup(group=group, role=role), h! h. H& E1 z' ]
return res9 Z5 p6 @1 x5 K8 K( {; v% l
def _creator_ur(user, role):
9 L7 a' U1 e/ i res = UserGroup(user=user, role=role)# C9 S" Q4 a* j; I# m5 ]
return res
' ?5 {& _+ F( n7 _- R- K##############################################################################
4 L, J% n, C$ @* M g: t# Object Model7 X8 ]) a+ ^9 O) g E$ {5 F& z, R) y
##############################################################################
! F: Q, k2 m' ~% x8 L% B, A. N& Gclass Role(Base, _BaseMixin):
! _* I& v/ n6 d; v __tablename__ = 'roles'- W: F1 X3 X/ S" P1 _: q, B
# columns* h6 e4 B( {0 f( F/ K. N+ \: E/ G
id = Column(Integer, primary_key=True, autoincrement=True)
) Z* i& H+ }% L9 P# K& ^! O. l name = Column(Unicode(16), unique=True)& d8 }3 ?2 D1 Z2 @" D8 h! G! x
# relations( a1 z' S+ t& F' z& t# L+ s. ?8 K8 P$ U
usergroup = relationship("UserGroup", backref='role')" \: A- K5 K$ z; ?( u% r d& k. E, M
class User(Base, _BaseMixin): h* \7 s9 ]# ~
__tablename__ = 'users'
- ]7 a3 X1 W4 B, y3 r # columns
8 I c ]6 y/ _5 K id = Column(Integer, primary_key=True, autoincrement=True)
& _1 h& ^+ Z# ~* o _ name = Column(Unicode(16), unique=True)
7 k8 e0 m2 P* w0 A # relations
+ y: X7 X: P" ^1 S _rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('group'),! K- d) O- I v8 g- r
cascade='all,delete-orphan',
4 L8 c1 k7 o; A2 K! A backref='user',( H; S" C7 ?. f! A& d5 Q
)/ p1 E$ t" N8 n( A
groups = association_proxy('_rel_usergroup', 'role', creator=_creator_gr)# E# v" h7 @, v8 Q6 E9 w
class Group(Base, _BaseMixin):" Y, C7 f2 K; ~
__tablename__ = 'groups'
) B8 T, H! m/ O) l) A # columns
8 b. T# l/ Z& t id = Column(Integer, primary_key=True, autoincrement=True)
, U+ G3 N4 n$ ~1 { j. F name = Column(Unicode(16), unique=True)+ F2 K a% m+ X G! B% u
# relations5 ~; y# b5 N( W" |' U6 g
_rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('user'),
& _( M- E# q( z4 r cascade='all,delete-orphan',+ M0 `8 P4 h9 A' f/ a! Q0 g- a
backref='group', i- @. n4 ?2 Z8 L# p6 R( l
)5 F4 ], y8 N" Q- F6 }
users = association_proxy('_rel_usergroup', 'role', creator=_creator_ur)
6 A9 S/ [, D1 S) s7 c1 W. aclass UserGroup(Base, _BaseMixin):
2 j: |& p& a0 J/ k __tablename__ = 'user_group'
& C1 J. X- c" i/ k # columns/ |3 D7 b, k9 K: H# ^0 V
id = Column(Integer, primary_key=True, autoincrement=True)" K. Y% q, F# I6 l8 u& i9 J8 m
user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False)8 X, c8 `9 }! ?" h
group_id = Column(Integer, ForeignKey('groups.id', ondelete='CASCADE'), nullable=False)$ A- V/ P$ V; R0 s
role_id = Column(Integer, ForeignKey('roles.id', ondelete='CASCADE'), nullable=False)5 ]5 z& S% `. L$ A! [. }2 \3 u
# relations
" ^9 t7 O }5 a # (all backrefs): I w" Y: U7 {
6 ?6 P/ K' D5 e, q3 s/ A5 ]
##############################################################################
" g- C2 ~, \* {5 I# TESTS (showing usages)+ U6 r' ^/ g7 Z9 c4 n% ^- a3 G
#5 T# y/ k5 o2 K7 w( g( r
# Requirements:% T' e F0 r# P6 \, A0 q6 {
# - list all groups of the user: user.groups (use keys)
2 ]# J% q- P+ M9 h# - list all users of the group: group.users (use keys)/ N- z0 w- w* B1 a, L! Z: L7 b
# - get all users ordered (grouped) by group with the role title
: a2 t6 g$ ~3 p8 K##############################################################################
$ a/ I( b0 W5 l7 S% w% ?def _requirement_get_user_groups(user):
( E W1 e3 w* N: ~8 V$ a$ ?8 ^ return user.groups.keys()
/ g0 t9 [2 {0 h$ w5 Cdef _requirement_get_group_users(group):
0 T# g/ W# [7 O5 d1 c' C" t return group.users.keys()
4 U2 W. A9 S- `' L; Y Rdef _requirement_get_all_users_by_group_with_role():# f& P, }8 ` ]: j
qry = session.query(Group).order_by(Group.name)
6 N& f% E1 i0 ? res = []
! y- M1 R4 W4 S, E& f$ r# c for g in qry.all():6 c7 \$ k) V2 j e, T" B
for u, r in sorted(g.users.items()):
/ ^" t- Z) y/ W6 P, V/ T value = (g.name, u.name, r.name)8 p0 ]0 I# [0 M/ G+ U
res.append(value)
e2 E" O6 ^. F1 a% \: P return res' x# g& X! B. S4 \6 d
def _test_all_requirements():
8 S; T0 c& x& K print '--requirement: all-ordered:'' ?" }8 |4 G9 Z3 \7 ?
for v in _requirement_get_all_users_by_group_with_role():
7 M8 ]+ b" u- y; a5 G5 d5 L print v
U w# }2 T3 [& a. i print '--requirement: user-groups:'9 v& K' n8 m9 S! i
for v in session.query(User).order_by(User.id):; r0 |' }9 u! A& @ p
print v, " has groups: ", _requirement_get_user_groups(v)0 H9 e$ c9 `, v( V3 e/ @; K- u
print '--requirement: group-users:'
% w# `: q) x9 n' D for v in session.query(Group).order_by(Group.id):0 s' L z" @" ]+ C; X8 @6 `8 J% W
print v, " has users: ", _requirement_get_group_users(v)& l6 u8 j0 J8 _" v
# create db schema* {$ d4 u1 f7 o/ t
Base.metadata.create_all(engine)7 ~" L4 G! c# B7 E( a7 o3 D
##############################################################################
* R) d5 D7 W2 c# a9 k$ R( {2 \# CREATE TEST DATA
( O$ g/ p0 N, L5 A4 n5 D; j' m) M4 d##############################################################################
0 X3 \& c8 _1 b# create entities
+ ^2 M1 y% K; H$ z# o# d9 @" m2 qu_peter = User(name='u_Peter')
4 z7 v" Q8 [- l+ U9 ?5 ?. |u_sonja = User(name='u_Sonja')& Q! ~4 R4 ~6 `+ F3 t8 ~
g_sales = Group(name='g_Sales')' K3 a7 E9 p+ B o: ]4 {; d y# X% @
g_wales = Group(name='g_Wales')
3 {* r# T! x. L- N! t1 I7 O$ or_super = Role(name='r_Super')5 f) G+ |- B1 s% g
r_minor = Role(name='r_Minor')
* M7 X1 h a b$ x# helper functions7 s2 g8 C! w( s/ C/ B
def _get_entity(entity, name):* ^9 m- N) u% p' Q
return session.query(entity).filter_by(name=name).one()
+ {( e, v8 I4 q7 v( W- qdef get_user(name):
! ~' I2 M7 J7 k7 ?7 D return _get_entity(User, name)
( F, j! V( k. e& ?- q& }# f- jdef get_group(name):% I3 B' ?; I, L9 O
return _get_entity(Group, name)
& u% U! n8 g! }+ odef _checkpoint():9 w6 P( P, F: E+ f* p+ |" U
session.commit()
4 b* s8 V2 f4 m& N8 \ session.expunge_all()9 c# L' ?$ m1 P- `. S- q {9 }
_test_all_requirements()
) Q, ^6 T, S2 t session.expunge_all()( K; N$ ]' k: G
print '-' * 808 i1 k( C4 \8 s& K: o
5 d1 @* `9 ]: m: E* I$ |+ _7 d# test: **ADD**
s* }1 k" h5 x; _) x5 ~u_peter.groups[g_wales] = r_minor # add9 v7 b U9 V0 y/ c# L1 P3 `/ J: b
g_wales.users[u_sonja] = r_super # add
6 Y8 r" T6 @' s9 R" k" {) Qg_sales.users[u_peter] = r_minor # add
}$ K( |5 j, \0 s& O$ v% P# Nsession.add(g_wales)
% \& b5 D# _2 ]. n2 O v: {2 R#session.add(g_sales)& B4 x, ]& ^0 D
_checkpoint(); H. G( J) ?) h+ G, j! Q. ], e: X; {1 R: |
# test: **UPDATE**
- A' F6 L$ y( M2 ~; e% e% Hu_peter = get_user('u_Peter')
E. i4 u; P M: f$ N' w. Z; D( s/ v9 hassert u_peter.name == 'u_Peter' and len(u_peter.groups) == 27 s' D% a# w4 n0 o. T+ }4 g% c
assert len(u_peter.groups) == 2, t, L1 R: r0 v5 M
g_wales = get_group('g_Wales')' F1 W5 C4 P _
g_wales.users[u_peter] = r_super # update& T# u% m: S( r s- f
_checkpoint()
; Z" X& K( B0 s+ ~0 p# test: **DELETE**
& F/ J$ M/ s- G- z# su_peter = get_user('u_Peter'), s4 F- R% w! A) T
assert u_peter.name == 'u_Peter' and len(u_peter.groups) == 2. |4 k/ O" a Y* ^% S' p
g_wales = get_group('g_Wales')
# n& F4 c! I* x- ydel u_peter.groups[g_wales] # delete
9 S' k0 z7 Q: Z+ W+ Y+ K_checkpoint() |
|