Gnumed.business.gmStaff

1 # -*- coding: utf8 -*- 2 """GNUmed staff objects.""" 3 #============================================================ 4 __author__ = "K.Hilbert <Karsten.Hilbert@gmx.net>" 5 __license__ = "GPL" 6 7 # std lib 8 import sys 9 import logging 10 11 # GNUmed 12 if __name__ == '__main__': 13 sys.path.insert(0, '../../') 14 from Gnumed.pycommon import gmBusinessDBObject 15 from Gnumed.pycommon import gmPG2 16 from Gnumed.pycommon import gmNull 17 from Gnumed.pycommon import gmBorg 18 from Gnumed.pycommon import gmLog2 19 20 21 _log = logging.getLogger('gm.staff') 22 23 _map_gm_role2pg_group = { 24 u'public': 'gm-public', 25 u'staff': u'gm-staff', 26 u'doctor': u'gm-doctors' 27 } 28 29 #============================================================ 30 _SQL_fetch_staff_fields = u'SELECT *, _(role) AS l10n_role FROM dem.v_staff WHERE %s' 31

32 -class cStaff(gmBusinessDBObject.cBusinessDBObject):

33 _cmd_fetch_payload = _SQL_fetch_staff_fields % u"pk_staff = %s" 34 _cmds_store_payload = [ 35 u"""UPDATE dem.staff SET 36 short_alias = %(short_alias)s, 37 comment = gm.nullify_empty_string(%(comment)s), 38 is_active = %(is_active)s, 39 db_user = %(db_user)s 40 WHERE 41 pk = %(pk_staff)s 42 AND 43 xmin = %(xmin_staff)s 44 RETURNING 45 xmin AS xmin_staff""" 46 ] 47 _updatable_fields = ['short_alias', 'comment', 'is_active', 'db_user'] 48 #--------------------------------------------------------

49 - def __init__(self, aPK_obj=None, row=None):

50 # by default get staff corresponding to CURRENT_USER 51 if (aPK_obj is None) and (row is None): 52 #cmd = u"select *, _(role) AS l10n_role from dem.v_staff where " 53 cmd = _SQL_fetch_staff_fields % u"db_user = CURRENT_USER" 54 try: 55 rows, idx = gmPG2.run_ro_queries(queries = [{'cmd': cmd}], get_col_idx=True) 56 except: 57 _log.exception('cannot instantiate staff instance') 58 gmLog2.log_stack_trace() 59 raise ValueError('cannot instantiate staff instance for database account CURRENT_USER') 60 if len(rows) == 0: 61 raise ValueError('no staff record for database account CURRENT_USER') 62 row = { 63 'pk_field': 'pk_staff', 64 'idx': idx, 65 'data': rows[0] 66 } 67 gmBusinessDBObject.cBusinessDBObject.__init__(self, row = row) 68 else: 69 gmBusinessDBObject.cBusinessDBObject.__init__(self, aPK_obj = aPK_obj, row = row) 70 71 # are we SELF ? 72 self.__is_current_user = (gmPG2.get_current_user() == self._payload[self._idx['db_user']]) 73 74 self.__inbox = None

75 #--------------------------------------------------------

76 - def __setitem__(self, attribute, value):

77 if attribute == 'db_user': 78 if self.__is_current_user: 79 _log.debug('will not modify database account association of CURRENT_USER staff member') 80 return 81 gmBusinessDBObject.cBusinessDBObject.__setitem__(self, attribute, value)

82 #--------------------------------------------------------

83 - def _get_db_lang(self):

84 rows, idx = gmPG2.run_ro_queries ( 85 queries = [{ 86 'cmd': u'select i18n.get_curr_lang(%(usr)s)', 87 'args': {'usr': self._payload[self._idx['db_user']]} 88 }] 89 ) 90 return rows[0][0]

91

92 - def _set_db_lang(self, language):

93 if not gmPG2.set_user_language(language = language): 94 raise ValueError ( 95 u'Cannot set database language to [%s] for user [%s].' % (language, self._payload[self._idx['db_user']]) 96 ) 97 return

98 99 database_language = property(_get_db_lang, _set_db_lang) 100 #--------------------------------------------------------

101 - def _get_inbox(self):

102 if self.__inbox is None: 103 from Gnumed.business import gmProviderInbox 104 self.__inbox = gmProviderInbox.cProviderInbox(provider_id = self._payload[self._idx['pk_staff']]) 105 return self.__inbox

106

107 - def _set_inbox(self, inbox):

108 return

109 110 inbox = property(_get_inbox, _set_inbox) 111 #--------------------------------------------------------

112 - def _get_identity(self):

113 from Gnumed.business import gmPerson 114 return gmPerson.cIdentity(aPK_obj = self._payload[self._idx['pk_identity']])

115 116 identity = property(_get_identity, lambda x:x) 117 #--------------------------------------------------------

118 - def set_role(self, conn=None, role=None):

119 if role.strip() == self._payload[self._idx['role']]: 120 return True 121 122 cmd = u'SELECT gm.add_user_to_permission_group(%(usr)s::name, %(grp)s::name)' 123 args = { 124 'usr': self._payload[self._idx['db_user']], 125 'grp': _map_gm_role2pg_group[role.strip()] 126 } 127 rows, idx = gmPG2.run_rw_queries ( 128 link_obj = conn, 129 queries = [{'cmd': cmd, 'args': args}], 130 get_col_idx = False, 131 return_data = True, 132 end_tx = True 133 ) 134 if not rows[0][0]: 135 return False 136 self.refetch_payload() 137 return True

138 139 role = property(lambda x:x, set_role)

140 #============================================================

141 -def get_staff_list(active_only=False):

142 if active_only: 143 cmd = _SQL_fetch_staff_fields % u'is_active ORDER BY can_login DESC, short_alias ASC' 144 else: 145 cmd = _SQL_fetch_staff_fields % u'TRUE ORDER BY can_login DESC, is_active DESC, short_alias ASC' 146 rows, idx = gmPG2.run_ro_queries(queries = [{'cmd': cmd}], get_col_idx=True) 147 staff_list = [] 148 for row in rows: 149 obj_row = { 150 'idx': idx, 151 'data': row, 152 'pk_field': 'pk_staff' 153 } 154 staff_list.append(cStaff(row=obj_row)) 155 return staff_list

156 #------------------------------------------------------------

157 -def create_staff(conn=None, db_account=None, password=None, identity=None, short_alias=None):

158 args = { 159 'pg_usr': db_account, 160 'pwd': password, 161 'person_id': identity, 162 'sig': short_alias, 163 'gm_role_name': u'doctor' 164 } 165 166 queries = [ 167 {'cmd': u'SELECT gm.create_user(%(pg_usr)s, %(pwd)s)', 'args': args}, 168 {'cmd': u""" 169 INSERT INTO dem.staff 170 (fk_identity, fk_role, db_user, short_alias) 171 VALUES ( 172 %(person_id)s, 173 (SELECT pk FROM dem.staff_role WHERE name = %(gm_role_name)s), 174 %(pg_usr)s, 175 %(sig)s 176 )""", 177 'args': args 178 } 179 ] 180 181 try: 182 rows, idx = gmPG2.run_rw_queries(link_obj = conn, queries = queries, end_tx = True) 183 except gmPG2.dbapi.IntegrityError, e: 184 if e.pgcode == gmPG2.sql_error_codes.UNIQUE_VIOLATION: 185 msg = _( 186 'Cannot add GNUmed user.\n' 187 '\n' 188 'The database account [%s] is already listed as a\n' 189 'GNUmed user. There can only be one GNUmed user\n' 190 'for each database account.\n' 191 ) % db_account 192 return False, msg 193 raise 194 195 return True, None

196 #------------------------------------------------------------

197 -def delete_staff(conn=None, pk_staff=None):

198 queries = [{'cmd': u'DELETE FROM dem.staff WHERE pk = %(pk)s', 'args': {'pk': pk_staff}}] 199 try: 200 rows, idx = gmPG2.run_rw_queries(link_obj = conn, queries = queries, end_tx = True) 201 except gmPG2.dbapi.IntegrityError, e: 202 if e.pgcode == gmPG2.sql_error_codes.FOREIGN_KEY_VIOLATION: # 23503 foreign_key_violation 203 msg = _( 204 'Cannot delete GNUmed staff member because the\n' 205 'database still contains data linked to it.\n' 206 '\n' 207 'The account was deactivated instead.' 208 ) 209 deactivate_staff(conn = conn, pk_staff = pk_staff) 210 return False, msg 211 raise 212 213 return True, None

214 #------------------------------------------------------------

215 -def activate_staff(conn=None, pk_staff=None):

216 # 1) activate staff entry 217 staff = cStaff(aPK_obj = pk_staff) 218 staff['is_active'] = True 219 staff.save_payload(conn=conn) # FIXME: error handling 220 221 # 2) enable database account login 222 rowx, idx = gmPG2.run_rw_queries ( 223 link_obj = conn, 224 # password does not matter because PG account must already exist 225 queries = [{'cmd': u'select gm.create_user(%s, %s)', 'args': [staff['db_user'], 'flying wombat']}], 226 end_tx = True 227 ) 228 229 return True

230 #------------------------------------------------------------

231 -def deactivate_staff(conn=None, pk_staff=None):

232 233 # 1) inactivate staff entry 234 staff = cStaff(aPK_obj = pk_staff) 235 staff['is_active'] = False 236 staff.save_payload(conn = conn) # FIXME: error handling 237 238 # 2) disable database account login 239 rows, idx = gmPG2.run_rw_queries ( 240 link_obj = conn, 241 queries = [{'cmd': u'select gm.disable_user(%s)', 'args': [staff['db_user']]}], 242 end_tx = True 243 ) 244 245 return True

246 #============================================================

247 -def set_current_provider_to_logged_on_user():

248 gmCurrentProvider(provider = cStaff())

249 #============================================================

250 -class gmCurrentProvider(gmBorg.cBorg):

251 """Staff member Borg to hold currently logged on provider. 252 253 There may be many instances of this but they all share state. 254 """

255 - def __init__(self, provider=None):

256 """Change or get currently logged on provider. 257 258 provider: 259 * None: get copy of current instance 260 * cStaff instance: change logged on provider (role) 261 """ 262 # make sure we do have a provider pointer 263 try: 264 self.provider 265 except AttributeError: 266 self.provider = gmNull.cNull() 267 268 # user wants copy of currently logged on provider 269 if provider is None: 270 return None 271 272 # must be cStaff instance, then 273 if not isinstance(provider, cStaff): 274 raise ValueError, 'cannot set logged on provider to [%s], must be either None or cStaff instance' % str(provider) 275 276 # same ID, no change needed 277 if self.provider['pk_staff'] == provider['pk_staff']: 278 return None 279 280 # first invocation 281 if isinstance(self.provider, gmNull.cNull): 282 self.provider = provider 283 return None 284 285 # user wants different provider 286 raise ValueError, 'provider change [%s] -> [%s] not yet supported' % (self.provider['pk_staff'], provider['pk_staff'])

287 288 #--------------------------------------------------------

289 - def get_staff(self):

290 return self.provider

291 #-------------------------------------------------------- 292 # __getitem__ handling 293 #--------------------------------------------------------

294 - def __getitem__(self, aVar):

295 """Return any attribute if known how to retrieve it by proxy. 296 """ 297 return self.provider[aVar]

298 #-------------------------------------------------------- 299 # __s/getattr__ handling 300 #--------------------------------------------------------

301 - def __getattr__(self, attribute):

302 if attribute == 'provider': # so we can __init__ ourselves 303 raise AttributeError 304 if not isinstance(self.provider, gmNull.cNull): 305 return getattr(self.provider, attribute)

306 # raise AttributeError 307 #============================================================ 308 # main/testing 309 #============================================================ 310 if __name__ == '__main__': 311 312 if len(sys.argv) == 1: 313 sys.exit() 314 315 if sys.argv[1] != 'test': 316 sys.exit() 317 318 import datetime 319 from Gnumed.pycommon import gmI18N 320 from Gnumed.pycommon import gmDateTime 321 322 gmI18N.activate_locale() 323 gmI18N.install_domain() 324 gmDateTime.init() 325 326 #--------------------------------------------------------

327 - def test_staff():

328 staff = cStaff() 329 print staff 330 print staff.inbox 331 print staff.inbox.messages

332 #--------------------------------------------------------

333 - def test_current_provider():

334 staff = cStaff() 335 provider = gmCurrentProvider(provider = staff) 336 print provider 337 print provider.inbox 338 print provider.inbox.messages 339 print provider.database_language 340 tmp = provider.database_language 341 provider.database_language = None 342 print provider.database_language 343 provider.database_language = tmp 344 print provider.database_language

345 #-------------------------------------------------------- 346 test_staff() 347 #test_current_provider() 348 349 #============================================================ 350

Source Code for Module Gnumed.business.gmStaff