Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1#!/usr/bin/env python 

2 

3""" 

4camcops_server/cc_modules/cc_group.py 

5 

6=============================================================================== 

7 

8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com). 

9 

10 This file is part of CamCOPS. 

11 

12 CamCOPS is free software: you can redistribute it and/or modify 

13 it under the terms of the GNU General Public License as published by 

14 the Free Software Foundation, either version 3 of the License, or 

15 (at your option) any later version. 

16 

17 CamCOPS is distributed in the hope that it will be useful, 

18 but WITHOUT ANY WARRANTY; without even the implied warranty of 

19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

20 GNU General Public License for more details. 

21 

22 You should have received a copy of the GNU General Public License 

23 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>. 

24 

25=============================================================================== 

26 

27**Group definitions.** 

28 

29""" 

30 

31import logging 

32from typing import List, Optional, Set 

33 

34from cardinal_pythonlib.logs import BraceStyleAdapter 

35from cardinal_pythonlib.reprfunc import simple_repr 

36from cardinal_pythonlib.sqlalchemy.orm_inspect import gen_columns 

37from cardinal_pythonlib.sqlalchemy.orm_query import exists_orm 

38from sqlalchemy.ext.associationproxy import association_proxy 

39from sqlalchemy.orm import relationship, Session as SqlASession 

40from sqlalchemy.sql.schema import Column, ForeignKey, Table 

41from sqlalchemy.sql.sqltypes import Integer 

42 

43from camcops_server.cc_modules.cc_ipuse import IpUse 

44from camcops_server.cc_modules.cc_policy import TokenizedPolicy 

45from camcops_server.cc_modules.cc_sqla_coltypes import ( 

46 GroupDescriptionColType, 

47 GroupNameColType, 

48 IdPolicyColType, 

49) 

50from camcops_server.cc_modules.cc_sqlalchemy import Base 

51 

52log = BraceStyleAdapter(logging.getLogger(__name__)) 

53 

54 

55# ============================================================================= 

56# Group-to-group association table 

57# ============================================================================= 

58# A group can always see itself, but may also have permission to see others; 

59# see "Groups" in the CamCOPS documentation. 

60 

61# http://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#self-referential-many-to-many-relationship # noqa 

62group_group_table = Table( 

63 "_security_group_group", 

64 Base.metadata, 

65 Column("group_id", Integer, ForeignKey("_security_groups.id"), 

66 primary_key=True), 

67 Column("can_see_group_id", Integer, ForeignKey("_security_groups.id"), 

68 primary_key=True) 

69) 

70 

71 

72# ============================================================================= 

73# Group 

74# ============================================================================= 

75 

76class Group(Base): 

77 """ 

78 Represents a CamCOPS group. 

79 

80 See "Groups" in the CamCOPS documentation. 

81 """ 

82 __tablename__ = "_security_groups" 

83 

84 id = Column( 

85 "id", Integer, 

86 primary_key=True, autoincrement=True, index=True, 

87 comment="Group ID" 

88 ) 

89 name = Column( 

90 "name", GroupNameColType, 

91 nullable=False, index=True, unique=True, 

92 comment="Group name" 

93 ) 

94 description = Column( 

95 "description", GroupDescriptionColType, 

96 comment="Description of the group" 

97 ) 

98 upload_policy = Column( 

99 "upload_policy", IdPolicyColType, 

100 comment="Upload policy for the group, as a string" 

101 ) 

102 finalize_policy = Column( 

103 "finalize_policy", IdPolicyColType, 

104 comment="Finalize policy for the group, as a string" 

105 ) 

106 

107 ip_use_id = Column( 

108 "ip_use_id", Integer, ForeignKey(IpUse.id), 

109 nullable=True, 

110 comment=f"FK to {IpUse.__tablename__}.{IpUse.id.name}" 

111 ) 

112 

113 ip_use = relationship(IpUse, uselist=False, single_parent=True, 

114 cascade="all, delete-orphan") 

115 

116 # users = relationship( 

117 # "User", # defined with string to avoid circular import 

118 # secondary=user_group_table, # link via this mapping table 

119 # back_populates="groups" # see User.groups 

120 # ) 

121 user_group_memberships = relationship( 

122 "UserGroupMembership", back_populates="group") 

123 users = association_proxy("user_group_memberships", "user") 

124 

125 regular_user_group_memberships = relationship( 

126 "UserGroupMembership", 

127 primaryjoin="and_(" 

128 "Group.id==UserGroupMembership.group_id, " 

129 "User.id==UserGroupMembership.user_id, " 

130 "User.auto_generated==False)" 

131 ) 

132 regular_users = association_proxy( 

133 "regular_user_group_memberships", 

134 "user" 

135 ) 

136 

137 can_see_other_groups = relationship( 

138 "Group", # link back to our own class 

139 secondary=group_group_table, # via this mapping table 

140 primaryjoin=(id == group_group_table.c.group_id), # "us" 

141 secondaryjoin=(id == group_group_table.c.can_see_group_id), # "them" 

142 backref="groups_that_can_see_us", 

143 lazy="joined" # not sure this does anything here 

144 ) 

145 

146 def __str__(self) -> str: 

147 return f"Group {self.id} ({self.name})" 

148 

149 def __repr__(self) -> str: 

150 attrnames = sorted(attrname for attrname, _ in gen_columns(self)) 

151 return simple_repr(self, attrnames) 

152 

153 def ids_of_other_groups_group_may_see(self) -> Set[int]: 

154 """ 

155 Returns a list of group IDs for groups that this group has permission 

156 to see. (Always includes our own group number.) 

157 """ 

158 group_ids = set() # type: Set[int] 

159 for other_group in self.can_see_other_groups: # type: Group 

160 other_group_id = other_group.id # type: Optional[int] 

161 if other_group_id is not None: 

162 group_ids.add(other_group_id) 

163 return group_ids 

164 

165 def ids_of_groups_group_may_see(self) -> Set[int]: 

166 """ 

167 Returns a list of group IDs for groups that this group has permission 

168 to see. (Always includes our own group number.) 

169 """ 

170 ourself = {self.id} # type: Set[int] 

171 return ourself.union(self.ids_of_other_groups_group_may_see()) 

172 

173 @classmethod 

174 def get_groups_from_id_list(cls, dbsession: SqlASession, 

175 group_ids: List[int]) -> List["Group"]: 

176 """ 

177 Fetches groups from a list of group IDs. 

178 """ 

179 return dbsession.query(Group).filter(Group.id.in_(group_ids)).all() 

180 

181 @classmethod 

182 def get_group_by_name(cls, dbsession: SqlASession, 

183 name: str) -> Optional["Group"]: 

184 """ 

185 Fetches a group from its name. 

186 """ 

187 if not name: 

188 return None 

189 return dbsession.query(cls).filter(cls.name == name).first() 

190 

191 @classmethod 

192 def get_group_by_id(cls, dbsession: SqlASession, 

193 group_id: int) -> Optional["Group"]: 

194 """ 

195 Fetches a group from its integer ID. 

196 """ 

197 if group_id is None: 

198 return None 

199 return dbsession.query(cls).filter(cls.id == group_id).first() 

200 

201 @classmethod 

202 def get_all_groups(cls, dbsession: SqlASession) -> List["Group"]: 

203 """ 

204 Returns all groups. 

205 """ 

206 return dbsession.query(Group).all() 

207 

208 @classmethod 

209 def all_group_ids(cls, dbsession: SqlASession) -> List[int]: 

210 """ 

211 Returns all group IDs. 

212 """ 

213 query = dbsession.query(cls).order_by(cls.id) 

214 return [g.id for g in query] 

215 

216 @classmethod 

217 def all_group_names(cls, dbsession: SqlASession) -> List[str]: 

218 """ 

219 Returns all group names. 

220 """ 

221 query = dbsession.query(cls).order_by(cls.id) 

222 return [g.name for g in query] 

223 

224 @classmethod 

225 def group_exists(cls, dbsession: SqlASession, group_id: int) -> bool: 

226 """ 

227 Does a particular group (specified by its integer ID) exist? 

228 """ 

229 return exists_orm(dbsession, cls, cls.id == group_id) 

230 

231 def tokenized_upload_policy(self) -> TokenizedPolicy: 

232 """ 

233 Returns the upload policy for a group. 

234 """ 

235 return TokenizedPolicy(self.upload_policy) 

236 

237 def tokenized_finalize_policy(self) -> TokenizedPolicy: 

238 """ 

239 Returns the finalize policy for a group. 

240 """ 

241 return TokenizedPolicy(self.finalize_policy)