回答

收藏

用户,组和角色的多对多声明式SQLAlchemy定义

技术问答 技术问答 241 人阅读 | 0 人回复 | 2023-09-14

我是SQLAlchemy的新手,我想知道定义此类表和关系的最佳方法是什么。我希望能够从,,,组中的用户访问user.groups用户的组group.users,并找出用户在组中的角色(我假设这将是在关联模型中定义的逻辑)。我还希望按组分组选择所有用户,并包括角色标题。
5 p$ X7 U' s# s/ s. L我已经使用教程创建关联表(声明样式),并删除试图secondary参数relationship的属性User和Group类别,但随后我就失去了能力,直接从用户的访问组,并直接从团体用户(backref的在关联表中指向关联类本身:/)。/ i4 j" `% \; i( g7 O
任何帮助,将不胜感激。
  u6 r$ O, v  Z( e% z$ \               
  U- v) [( u1 ?1 k! i1 h解决方案:! s' e7 n" j& y, W% b" _- a
               
  [  Z/ i! W% z+ g  o5 U3 w* g
! q. L) Q8 S8 ]$ X+ E2 U1 w  ^/ p4 j( P) J* t2 r. M. f# n
                简化模型的关键是使用associationproxy,因此您一定要检查一下。' m4 }1 W9 ?/ G* ^. c( E
假设 用户 在一个组中只能有一个 角色 ,则下面的代码应回答您的所有问题:5 Q7 U( r/ [  Q9 K
如何配置模型和关系. a. h" m6 ]8 Y' [# |5 [5 L
如何添加/删除/更新角色- Q" b% t2 H& n
如何检索您要求的数据(报告)
- y" W% B# J& i, I. F, a
/ r  O  R/ D) n* H* @
您应该接管模型部分,而忽略其余部分。下面是完全包含且有效的脚本:4 S$ v- i% u/ f- u' ]7 A
from sqlalchemy import create_engine, Column, Integer, Unicode, ForeignKey5 }  I( B# I5 o8 i$ O# B, }) ]
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
1 b& Q9 J7 T3 o( Q' @from sqlalchemy.orm.collections import attribute_mapped_collection$ W& J, R/ @% }/ H4 P# i4 r3 j
from sqlalchemy.ext.declarative import declarative_base
6 x& }% G9 a' i1 Q! gfrom sqlalchemy.ext.associationproxy import association_proxy: p6 j: S- k) ?1 q# f* u5 o
# Configure test data SA
$ {. `7 z  K0 \4 q' ^engine = create_engine(u'sqlite:///:memory:', echo=False)
8 h( C; j+ e7 S8 o: ksession = scoped_session(sessionmaker(bind=engine, autoflush=False))" K/ D- Q: j8 g6 \
Base = declarative_base(), k" t0 V: M* H* C# C
class _BaseMixin(object):/ x4 s6 ?4 k+ k1 E7 N1 J3 a1 h
    """' f; F. w' j5 ~8 n# d
    A helper mixin class to set properties on object creation.: y+ V0 z! r2 U
    Also provides a convenient default __repr__() function, but be aware that
/ S( U. \3 V- h4 I9 [5 X4 M    also relationships are printed, which might result in loading the relation' C7 M/ n/ R# T: ?( e
    objects from the database' r: e  p9 K% p) q3 F) L* |
    """$ `' {2 Y  v. a0 `2 W& n4 ]
    def __init__(self, **kwargs):7 y8 P& N3 l7 _9 V: Z/ E% r' s, z/ O
        for k, v in kwargs.items():
. H3 w2 e; {9 [8 R1 w8 G            setattr(self, k, v)' B  N1 R  }! t! e
    def __repr__(self):
& v% q: U$ d8 j' {, F; n. K& Q4 h        return "" % (self.__class__.__name__,
# |0 H  A# D8 V' F: |/ V            ', '.join('%s=%r' % (k, self.__dict__[k])
% c. @% r2 ?/ x, t0 O! f+ f                      for k in sorted(self.__dict__)# Q  G8 f( Z; `6 _1 l3 q
                      if '_' != k[0]
. N  w: V0 Y0 X  ~$ |3 u) ~5 X4 _                      #if '_sa_' != k[:4] and '_backref_' != k[:9]  _% r' `9 h8 I7 ~; c$ t3 L
                      )
9 j' U) p$ Q2 i  N            )
0 ?5 V; W  M& R  s  p. [
- N8 p( f# A5 S& j6 L, m  W1 L7 q# relation creator factory functions# n* s! c- F8 x7 j0 ~
def _creator_gr(group, role):
, i; s9 L) l2 a1 \; o7 y2 S' U, Y    res = UserGroup(group=group, role=role)7 k, q$ G8 m+ y# N" Q9 d
    return res
, v' R1 ~! f4 \0 Z% j8 xdef _creator_ur(user, role):% H1 I5 T0 v* m& T0 x
    res = UserGroup(user=user, role=role)
& r4 C9 \' p; a' C' h$ r4 N    return res+ r; k5 S4 V8 C9 y8 o3 ]0 w  Z2 B7 C
##############################################################################: X' Z$ B2 z- h5 C" Z3 l
# Object Model% {) q/ t% n6 @
##############################################################################
% L6 k9 S8 ^) U) c- }7 Hclass Role(Base, _BaseMixin):" X0 H, J, C  g
    __tablename__ = 'roles'* g$ X6 g5 b! U/ n+ @" ~: a7 x  p# F0 q
    # columns
3 Q2 v5 [  Q! J( c' M* F2 `3 p    id = Column(Integer, primary_key=True, autoincrement=True)
& Q( m% F+ `2 x+ F. w, M) z+ k9 r    name = Column(Unicode(16), unique=True)- D6 P4 E4 C7 o+ Y! |. W  E# `4 |
    # relations
$ A  b" f3 w0 U    usergroup = relationship("UserGroup", backref='role')
7 j- t/ @, z  [9 ~7 \( b5 bclass User(Base, _BaseMixin):
5 w$ l1 j- h+ Q( V7 }1 e& u    __tablename__ = 'users'5 ~: S# a# l- N
    # columns
, c! H7 f# S9 X0 ]" u8 B    id = Column(Integer, primary_key=True, autoincrement=True)
5 Z8 {( p* J% S/ S0 t" G' Y0 ?) E+ @% I    name = Column(Unicode(16), unique=True)+ i. s0 T7 F" m, Z5 J
    # relations8 r# l) f+ l6 i0 b2 A/ v& A
    _rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('group')," O: m7 [5 g+ F2 p( B! {- U. ?/ y
                                  cascade='all,delete-orphan',
0 o& X. k( I* _) s9 k                                  backref='user',6 q! ~4 T# ^1 Z; Y# M0 B2 M* U" M
                                  )3 G7 B8 }0 h5 c* T. v
    groups = association_proxy('_rel_usergroup', 'role', creator=_creator_gr)
# L* K, [9 b8 _0 P/ T4 g) Lclass Group(Base, _BaseMixin):
% j' j; h! V! v$ H$ v    __tablename__ = 'groups'( @; P! L9 Z0 E' K% ^& D; d
    # columns& N+ b- `( j4 t' |* C
    id = Column(Integer, primary_key=True, autoincrement=True)
% j0 B; y" X1 k& X/ V( Z    name = Column(Unicode(16), unique=True)' e$ u: \. a. G' w, l
    # relations
- N$ n- c( ?0 @6 R9 e, I    _rel_usergroup = relationship("UserGroup", collection_class=attribute_mapped_collection('user'),2 k- _4 D3 g/ F
                                  cascade='all,delete-orphan',
+ r' ?, S+ F/ p, H& h                                  backref='group',
+ D8 [1 Q' S% e                                  )) U% |3 M, B/ E: |/ V8 ]* i( m
    users = association_proxy('_rel_usergroup', 'role', creator=_creator_ur)
/ a: B- Y( ]1 X; L$ D# fclass UserGroup(Base, _BaseMixin):7 E$ J1 M  l: v  b: i
    __tablename__ = 'user_group'
5 {: `' v- w* y    # columns6 G3 h  M, n! R6 z. [, z
    id = Column(Integer, primary_key=True, autoincrement=True)0 b: B2 E" ~- C9 f
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False)) {& ~6 a; o1 o- |& {
    group_id = Column(Integer, ForeignKey('groups.id', ondelete='CASCADE'), nullable=False)
3 m+ c3 |2 A/ V8 e2 C. \- n$ o  T    role_id = Column(Integer, ForeignKey('roles.id', ondelete='CASCADE'), nullable=False)
$ C+ i, T, M7 H' q  @2 ^    # relations$ a, g2 H+ {- }/ i
    # (all backrefs)6 ?+ ~; [4 h: H' l; d8 t' o% R3 C  ~

. r9 I# [5 ]4 W  B. Q  }* O# n; R##############################################################################
" ?2 l" D7 H% r! W: k3 x# TESTS (showing usages)
  E. p/ i# k! C- s, ^$ ~#; z4 {7 J! m% |% X
# Requirements:
4 R1 H/ Z. c  [+ r#  - list all groups of the user: user.groups (use keys)
$ Y1 t. i9 `0 ^. i" O( @#  - list all users of the group: group.users (use keys)
: m) y1 p( `# a% J5 f) j7 u#  - get all users ordered (grouped) by group with the role title0 n# Z# ~4 [% [1 |0 c( R* }
##############################################################################* S+ b; H4 C; [6 `, ]
def _requirement_get_user_groups(user):3 s& {# I' C9 C9 p; K# {6 G
    return user.groups.keys()8 r/ r# E: l0 w# h5 i
def _requirement_get_group_users(group):
% s' ^  Y& R' ^" f    return group.users.keys()
- Y4 ?7 `9 N9 [2 ~: jdef _requirement_get_all_users_by_group_with_role():
) M, e7 Z/ k- Q- ?    qry = session.query(Group).order_by(Group.name)8 \4 n* S5 N5 C. Y" L: `
    res = []% u* v7 l7 J' H% g3 Z
    for g in qry.all():; w% |7 @  D7 Q" M2 n: k3 M
        for u, r in sorted(g.users.items()):
8 t; K* o6 Y6 d" C# {$ `# q            value = (g.name, u.name, r.name)
, g4 c. S& r6 H! k7 O9 o            res.append(value)2 N5 \+ r* e: y! B
    return res/ n0 T1 h8 s/ W( N% [- U0 D4 W5 p2 I
def _test_all_requirements():4 M, E% ~% c1 g# G: [& F
    print '--requirement: all-ordered:'
- ~1 {3 n. C5 @9 \( \$ b& t    for v in _requirement_get_all_users_by_group_with_role():" v$ b/ t6 q9 Q& e$ K
        print v- }2 w$ [& n; _+ o5 [5 v9 `! |
    print '--requirement: user-groups:'
  y! D+ o* R2 D, u8 v, ^    for v in session.query(User).order_by(User.id):
( G  H& C, u1 }        print v, " has groups: ",  _requirement_get_user_groups(v)
$ s5 Z+ V+ f, U, {, S: V) z) `    print '--requirement: group-users:'
+ @6 |: O3 h; d4 z( R* r    for v in session.query(Group).order_by(Group.id):( \: N" x+ {/ {" d3 c2 t+ W
        print v, " has users: ",  _requirement_get_group_users(v)& R: E% d" f- a; E& x, e, G& i
# create db schema
6 q5 M% C% k2 p* x/ m& rBase.metadata.create_all(engine)' I, H9 q" q8 J" ^8 ~5 e5 f8 r+ @
##############################################################################4 N( e3 L/ |+ P) A1 A
# CREATE TEST DATA' `) W' Y' N) K: H- Y
##############################################################################6 a+ {6 x! ?; y4 ^
# create entities! L: k* m: ~  |7 L
u_peter = User(name='u_Peter')
" Y# K) J% H( F( I: N+ s- Xu_sonja = User(name='u_Sonja')( [* c3 o. s$ }. M3 R: X5 B* i
g_sales = Group(name='g_Sales')
! }& x  T  X9 C% ig_wales = Group(name='g_Wales')
; Y) a# U+ E/ Y8 N; Cr_super = Role(name='r_Super')* n5 h* A- j3 v" ^) A( `- Y
r_minor = Role(name='r_Minor')
! v8 l- ?" V* W; S* x' ^# helper functions
2 H% t( J3 P- ^4 Rdef _get_entity(entity, name):1 B1 G6 \$ l) K! B6 I5 s0 e
    return session.query(entity).filter_by(name=name).one()
% W. a' M  I2 @$ P- t2 ]def get_user(name):# a* _8 F4 d4 M3 q5 m% a5 S( d
    return _get_entity(User, name)
5 z; I5 B* L: k) \- Y* jdef get_group(name):" r# Y9 g. c& i! n# m- {& [8 ~
    return _get_entity(Group, name)! c9 O8 R& X% D/ K. U
def _checkpoint():: ?1 ^4 ]% R3 e. u" `: ^
    session.commit()
8 e% b# ]# P- I+ y, b    session.expunge_all()
" V8 z& u0 j: P) n7 W. C    _test_all_requirements()
& g: b; D, l2 E5 [+ W5 |* K: A. q    session.expunge_all()
4 {+ o& {, V: Z, P' a    print '-' * 80' N9 Q' F3 ~0 c4 ~1 f

5 H* N. R1 T! R9 y6 C( t: {% z, H# test: **ADD**
- S1 W+ f6 U' [+ i; R+ ]" X! c% ?5 Wu_peter.groups[g_wales] = r_minor # add
1 O6 D$ p2 l3 g7 ag_wales.users[u_sonja] = r_super # add
& i3 B3 P9 _" _7 g: _g_sales.users[u_peter] = r_minor # add, w9 H8 R. y1 i
session.add(g_wales)
; }) Y" n: }4 ]- _6 j#session.add(g_sales)# c  S9 b$ m4 P7 ~" j
_checkpoint()( @8 L" Y, @( ~% I: O4 a2 K, k
# test: **UPDATE**
/ N% q  n' `2 W7 y: O0 s7 fu_peter = get_user('u_Peter')5 b, @* h3 A! @) V$ s- [
assert u_peter.name == 'u_Peter' and len(u_peter.groups) == 2& v: a/ p$ X$ L! y4 a  t
assert len(u_peter.groups) == 2" S2 X9 m* w  _  ^% S
g_wales = get_group('g_Wales')
* I% {$ M8 ]; L+ C" X4 Gg_wales.users[u_peter] = r_super # update
5 E8 n# L: l/ \+ k0 O- j_checkpoint()  o7 j+ B+ w- W: j% D
# test: **DELETE**
% N& y5 [% Y: z( q) l6 A* zu_peter = get_user('u_Peter')
* m; I; v; x: ?+ ]! \, hassert u_peter.name == 'u_Peter' and len(u_peter.groups) == 2
+ n, f# z6 L; \9 L- H) [g_wales = get_group('g_Wales')
2 }$ [* }2 S, U3 x7 o9 Pdel u_peter.groups[g_wales] # delete
3 I1 h, [* t/ R7 m1 i_checkpoint()
分享到:
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则