Module Gnumed.business.gmHospitalStay
GNUmed hospital stay business object.
license: GPL v2 or later
Expand source code
# -*- coding: utf-8 -*-
"""GNUmed hospital stay business object.
license: GPL v2 or later
"""
#============================================================
__author__ = "<karsten.hilbert@gmx.net>"
import sys
import logging
if __name__ == '__main__':
sys.path.insert(0, '../../')
_ = lambda x:x
from Gnumed.pycommon import gmPG2
from Gnumed.pycommon import gmI18N
from Gnumed.pycommon import gmTools
from Gnumed.pycommon import gmBusinessDBObject
from Gnumed.business import gmDocuments
_log = logging.getLogger('gm.emr')
#============================================================
_SQL_get_hospital_stays = "select * from clin.v_hospital_stays where %s"
class cHospitalStay(gmBusinessDBObject.cBusinessDBObject):
_cmd_fetch_payload = _SQL_get_hospital_stays % "pk_hospital_stay = %s"
_cmds_store_payload = [
"""UPDATE clin.hospital_stay SET
clin_when = %(admission)s,
discharge = %(discharge)s,
fk_org_unit = %(pk_org_unit)s,
narrative = gm.nullify_empty_string(%(comment)s),
fk_episode = %(pk_episode)s,
fk_encounter = %(pk_encounter)s
WHERE
pk = %(pk_hospital_stay)s
AND
xmin = %(xmin_hospital_stay)s
RETURNING
xmin AS xmin_hospital_stay
"""
]
_updatable_fields = [
'admission',
'discharge',
'pk_org_unit',
'pk_episode',
'pk_encounter',
'comment'
]
#--------------------------------------------------------
def format_maximum_information(self, patient=None):
return self.format (
include_procedures = True,
include_docs = True
).split('\n')
#-------------------------------------------------------
def format(self, left_margin=0, include_procedures=False, include_docs=False, include_episode=True):
if self._payload['discharge']:
discharge = ' - %s' % self._payload['discharge'].strftime('%Y %b %d')
else:
discharge = ''
episode = ''
if include_episode:
episode = ': %s%s%s' % (
gmTools.u_left_double_angle_quote,
self._payload['episode'],
gmTools.u_right_double_angle_quote
)
lines = ['%s%s%s (%s@%s)%s' % (
' ' * left_margin,
self._payload['admission'].strftime('%Y %b %d'),
discharge,
self._payload['ward'],
self._payload['hospital'],
episode
)]
if include_docs:
for doc in self.documents:
lines.append('%s%s: %s\n' % (
' ' * left_margin,
_('Document'),
doc.format(single_line = True)
))
return '\n'.join(lines)
#--------------------------------------------------------
def _get_documents(self):
return [ gmDocuments.cDocument(aPK_obj = pk_doc) for pk_doc in self._payload['pk_documents'] ]
documents = property(_get_documents)
#-----------------------------------------------------------
def get_latest_patient_hospital_stay(patient:int=None) -> cHospitalStay | None:
"""Actually, the stay with the most recent admission."""
SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s ORDER BY admission DESC LIMIT 1"
queries = [{'sql': SQL, 'args': {'pat': patient}}]
rows = gmPG2.run_ro_queries(queries = queries)
if not rows:
return None
return cHospitalStay(row = {'data': rows[0], 'pk_field': 'pk_hospital_stay'})
#-----------------------------------------------------------
def get_patient_hospital_stays(patient:int=None, ongoing_only:bool=False, return_pks:bool=False) -> list[cHospitalStay]:
args = {'pat': patient}
if ongoing_only:
SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s AND discharge is NULL ORDER BY admission"
else:
SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s ORDER BY admission"
queries = [{'sql': SQL, 'args': args}]
rows = gmPG2.run_ro_queries(queries = queries)
if return_pks:
return [ r['pk_hospital_stay'] for r in rows ]
return [ cHospitalStay(row = {'data': r, 'pk_field': 'pk_hospital_stay'}) for r in rows ]
#-----------------------------------------------------------
def create_hospital_stay(encounter:int=None, episode:int=None, fk_org_unit:int=None) -> cHospitalStay:
SQL = 'INSERT INTO clin.hospital_stay (fk_encounter, fk_episode, fk_org_unit) VALUES (%(enc)s, %(epi)s, %(fk_org_unit)s) RETURNING pk'
args = {'enc': encounter, 'epi': episode, 'fk_org_unit': fk_org_unit}
queries = [{'sql': SQL, 'args': args}]
rows = gmPG2.run_rw_queries(queries = queries, return_data = True)
return cHospitalStay(aPK_obj = rows[0][0])
#-----------------------------------------------------------
def delete_hospital_stay(stay:int=None):
cmd = 'DELETE FROM clin.hospital_stay WHERE pk = %(pk)s'
args = {'pk': stay}
gmPG2.run_rw_queries(queries = [{'sql': cmd, 'args': args}])
return True
#============================================================
# main - unit testing
#------------------------------------------------------------
if __name__ == '__main__':
if len(sys.argv) < 2:
sys.exit()
if sys.argv[1] != 'test':
sys.exit()
gmI18N.activate_locale()
gmI18N.install_domain('gnumed')
#--------------------------------------------------------
# define tests
#--------------------------------------------------------
def test_hospital_stay():
stay = create_hospital_stay(encounter = 1, episode = 2, fk_org_unit = 1)
# stay['hospital'] = u'Starfleet Galaxy General Hospital'
# stay.save_payload()
print(stay)
for s in get_patient_hospital_stays(12):
print(s)
print(s.format())
delete_hospital_stay(stay['pk_hospital_stay'])
print('should fail:')
stay = create_hospital_stay(encounter = 1, episode = 4, fk_org_unit = 1)
#--------------------------------------------------------
gmPG2.request_login_params(setup_pool = True)
test_hospital_stay()Functions
def create_hospital_stay(encounter: int = None, episode: int = None, fk_org_unit: int = None) ‑> cHospitalStay-
Expand source code
def create_hospital_stay(encounter:int=None, episode:int=None, fk_org_unit:int=None) -> cHospitalStay: SQL = 'INSERT INTO clin.hospital_stay (fk_encounter, fk_episode, fk_org_unit) VALUES (%(enc)s, %(epi)s, %(fk_org_unit)s) RETURNING pk' args = {'enc': encounter, 'epi': episode, 'fk_org_unit': fk_org_unit} queries = [{'sql': SQL, 'args': args}] rows = gmPG2.run_rw_queries(queries = queries, return_data = True) return cHospitalStay(aPK_obj = rows[0][0]) def delete_hospital_stay(stay: int = None)-
Expand source code
def delete_hospital_stay(stay:int=None): cmd = 'DELETE FROM clin.hospital_stay WHERE pk = %(pk)s' args = {'pk': stay} gmPG2.run_rw_queries(queries = [{'sql': cmd, 'args': args}]) return True def get_latest_patient_hospital_stay(patient: int = None) ‑> cHospitalStay | None-
Actually, the stay with the most recent admission.
Expand source code
def get_latest_patient_hospital_stay(patient:int=None) -> cHospitalStay | None: """Actually, the stay with the most recent admission.""" SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s ORDER BY admission DESC LIMIT 1" queries = [{'sql': SQL, 'args': {'pat': patient}}] rows = gmPG2.run_ro_queries(queries = queries) if not rows: return None return cHospitalStay(row = {'data': rows[0], 'pk_field': 'pk_hospital_stay'}) def get_patient_hospital_stays(patient: int = None, ongoing_only: bool = False, return_pks: bool = False) ‑> list[cHospitalStay]-
Expand source code
def get_patient_hospital_stays(patient:int=None, ongoing_only:bool=False, return_pks:bool=False) -> list[cHospitalStay]: args = {'pat': patient} if ongoing_only: SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s AND discharge is NULL ORDER BY admission" else: SQL = _SQL_get_hospital_stays % "pk_patient = %(pat)s ORDER BY admission" queries = [{'sql': SQL, 'args': args}] rows = gmPG2.run_ro_queries(queries = queries) if return_pks: return [ r['pk_hospital_stay'] for r in rows ] return [ cHospitalStay(row = {'data': r, 'pk_field': 'pk_hospital_stay'}) for r in rows ]
Classes
class cHospitalStay (aPK_obj: int | dict = None, row: dict = None, link_obj=None)-
Represents business objects in the database.
Rules
- instances ARE ASSUMED TO EXIST in the database
- PK construction (aPK_obj): DOES verify its existence on instantiation (fetching data fails)
- Row construction (row): allowed by using a dict of pairs of field name: field value (PERFORMANCE improvement)
- does NOT verify FK target existence
- does NOT create new entries in the database
- does NOT lazy-fetch fields on access
Class scope SQL commands and variables:
_cmd_fetch_payload:
- must return exactly one row
- WHERE clause argument values are expected in self.pk_obj (taken from init(aPK_obj))
- must return xmin of all rows that _cmds_store_payload will be updating, so views must support the xmin columns of their underlying tables
_cmds_store_payload:
- one or multiple "update … set … where xmin_ = … and pk = …" statements which actually update the database from the data in self._payload,
- the last query must refetch at least the XMIN values needed to detect concurrent updates, their field names had better be the same as in _cmd_fetch_payload,
- the last query CAN return other fields which is particularly useful when those other fields are computed in the backend and may thus change upon save but will not have been set by the client code explicitly - this is only really of concern if the saved subclass is to be reused after saving rather than re-instantiated
- when subclasses tend to live a while after save_payload() was called and they support computed fields (say, _(some_column) you need to return all columns (see cEncounter)
_updatable_fields:
- a list of fields available for update via object['field']
Call init from child classes like so:
super().__init__(aPK_obj = aPK_obj, row = row, link_obj = link_obj)Args
aPK_obj- retrieve data from backend
- an scalar value the ._cmd_fetch_payload WHERE condition must be a simple column: "… WHERE pk_col = %s"
- a dictionary of values the ._cmd_fetch_payload WHERE condition must consume the dictionary and produce a unique row
row- must hold the fields
- data: list of column values for the row selected by ._cmd_fetch_payload (as returned by cursor.fetchone() in the DB-API)
- pk_field: the name of the primary key column OR
- pk_obj: a dictionary suitable for being passed to cursor.execute and holding the primary key values, used for composite PKs
- for example:
row = { 'data': rows[0], 'pk_field': 'pk_XXX (the PK column name)', 'pk_obj': {'pk_col1': pk_col1_val, 'pk_col2': pk_col2_val} } rows = gmPG2.run_ro_queries(queries = [{'sql': cmd, 'args': args}]) objects = [ cChildClass(row = {'data': r, 'pk_field': 'the PK column name'}) for r in rows ]
Expand source code
class cHospitalStay(gmBusinessDBObject.cBusinessDBObject): _cmd_fetch_payload = _SQL_get_hospital_stays % "pk_hospital_stay = %s" _cmds_store_payload = [ """UPDATE clin.hospital_stay SET clin_when = %(admission)s, discharge = %(discharge)s, fk_org_unit = %(pk_org_unit)s, narrative = gm.nullify_empty_string(%(comment)s), fk_episode = %(pk_episode)s, fk_encounter = %(pk_encounter)s WHERE pk = %(pk_hospital_stay)s AND xmin = %(xmin_hospital_stay)s RETURNING xmin AS xmin_hospital_stay """ ] _updatable_fields = [ 'admission', 'discharge', 'pk_org_unit', 'pk_episode', 'pk_encounter', 'comment' ] #-------------------------------------------------------- def format_maximum_information(self, patient=None): return self.format ( include_procedures = True, include_docs = True ).split('\n') #------------------------------------------------------- def format(self, left_margin=0, include_procedures=False, include_docs=False, include_episode=True): if self._payload['discharge']: discharge = ' - %s' % self._payload['discharge'].strftime('%Y %b %d') else: discharge = '' episode = '' if include_episode: episode = ': %s%s%s' % ( gmTools.u_left_double_angle_quote, self._payload['episode'], gmTools.u_right_double_angle_quote ) lines = ['%s%s%s (%s@%s)%s' % ( ' ' * left_margin, self._payload['admission'].strftime('%Y %b %d'), discharge, self._payload['ward'], self._payload['hospital'], episode )] if include_docs: for doc in self.documents: lines.append('%s%s: %s\n' % ( ' ' * left_margin, _('Document'), doc.format(single_line = True) )) return '\n'.join(lines) #-------------------------------------------------------- def _get_documents(self): return [ gmDocuments.cDocument(aPK_obj = pk_doc) for pk_doc in self._payload['pk_documents'] ] documents = property(_get_documents)Ancestors
Instance variables
var documents-
Expand source code
def _get_documents(self): return [ gmDocuments.cDocument(aPK_obj = pk_doc) for pk_doc in self._payload['pk_documents'] ]
Methods
def format_maximum_information(self, patient=None)-
Expand source code
def format_maximum_information(self, patient=None): return self.format ( include_procedures = True, include_docs = True ).split('\n')
Inherited members