Module Gnumed.business.gmPatientTransplant
GNUmed patient transplant.
Export all data belonging to one patient as a self-contained SQL script. Running that script against another GNUmed database of the same schema version puts the patient into that database ("transplant").
If the patient seems to already exist in the target database the script aborts (rolling back everything) and reports the candidate identities such that the user can decide whether to create a new patient or to add the data to an existing one.
Properties of the generated script:
-
plain SQL, runnable by psql as well as by implant_patient()
-
one single transaction, nothing is changed unless everything succeeds
-
a block near the top can be edited manually: staff matching mode, staff mapping, "same patient as #…", patient mode, force, dry run
-
all patient data is first loaded into constraint-free temporary staging tables, all subsequent statements do not contain patient data literals, such that server side error logging cannot leak patient data
-
primary keys are never transferred: the target allocates new ones, the source->target mapping of rows is persisted into dem.identity.aux_info->'gm_transplant' such that repeated imports (forced or delta) never duplicate rows
-
the original audit metadata (.modified_by/.modified_when) and, if available, the audit trail of the source are put into the target as documents of the patient
Functions
def explant_patient(pk_identity: int,
filename: str = None,
conn=None,
include_history: bool = True,
passphrase: str = None,
source_description: str = None,
progress_callback=None) ‑> str-
Expand source code
def explant_patient(pk_identity:int, filename:str=None, conn=None, include_history:bool=True, passphrase:str=None, source_description:str=None, progress_callback=None) -> str: """Export a patient as a transplant script. Contrary to what EXplant might suggest, source data is NOT removed from the source. Args: pk_identity: the patient (dem.identity.pk) filename: where to put the script, default: work directory conn: a connection to read from, should be gm-dbo such that the audit trail can be read; if None a connection of the current user is used include_history: whether to include the audit trail passphrase: if given, the script is encrypted symmetrically source_description: shown to the user importing the script Returns: The name of the (possibly encrypted) script file. Raises: cTransplantError: messages contain no patient data """ if filename is None: filename = gmTools.get_unique_filename ( prefix = 'gm-transplant-pk%s-%s-' % (pk_identity, pydt.datetime.now().strftime('%Y%m%d-%H%M%S')), suffix = '.sql', tmp_dir = gmTools.gmPaths().user_work_dir ) filename = os.path.abspath(os.path.expanduser(filename)) conn, close_conn = _get_conn(conn) _log.info('exporting patient #%s for transplant into [%s]', pk_identity, filename) curs = conn.cursor() try: problems = _verify_export_completeness(curs) finally: curs.close() if problems: close_conn() raise cTransplantError(_('This GNUmed version cannot fully transplant patients from this database:\n\n%s') % '\n'.join(problems)) schema_hash = gmPG2.get_schema_hash(link_obj = conn) expected_hash = gmPG2.known_schema_hashes.get(SCHEMA_VERSION) if schema_hash != expected_hash: close_conn() raise cTransplantError(_('Database schema hash [%s] does not match the expected v%s schema hash [%s].') % (schema_hash, SCHEMA_VERSION, expected_hash)) if source_description is None: source_description = _get_source_description() prev_autocommit = conn.autocommit try: conn.rollback() conn.autocommit = True exporter = _cExporter ( conn = conn, pk_identity = pk_identity, include_history = include_history, source_description = source_description, progress_callback = progress_callback ) exporter.schema_hash = schema_hash exporter.export(filename) except cTransplantError: _log.exception('patient #%s cannot be exported for transplant', pk_identity) raise except Exception as exc: _log.exception('error exporting patient #%s for transplant', pk_identity) raise cTransplantError(_('Error exporting the patient (%s). Check the log.') % exc.__class__.__name__) from exc finally: try: conn.autocommit = prev_autocommit except Exception: pass close_conn() _log.info('patient #%s exported: %s rows', pk_identity, sum(exporter.counts.values())) if passphrase: from Gnumed.pycommon import gmCrypto # GPG such that import_patient_from_transplant_script() can decrypt it encrypted = gmCrypto.encrypt_file_symmetric_gpg ( filename = filename, passphrase = passphrase, comment = 'GNUmed transplant script', remove_unencrypted = True ) if not encrypted: raise cTransplantError(_('Cannot encrypt transplant script (is GnuPG installed ?).')) os.chmod(encrypted, 0o600) return encrypted return filenameExport a patient as a transplant script.
Contrary to what EXplant might suggest, source data is NOT removed from the source.
- Args
- -----=
pk_identity- the patient (dem.identity.pk)
filename- where to put the script, default: work directory
conn- a connection to read from, should be gm-dbo such that the audit trail can be read; if None a connection of the current user is used
include_history- whether to include the audit trail
passphrase- if given, the script is encrypted symmetrically
source_description- shown to the user importing the script
Returns -----= The name of the (possibly encrypted) script file.
- Raises
- -----=
cTransplantError- messages contain no patient data
def implant_patient(filename: str,
conn,
patient_mode: str = None,
same_patient_as: int = None,
staff_match: str = None,
fallback_staff: str = None,
dry_run: bool = False,
force: bool = False,
progress_callback=None) ‑> cTransplantResult-
Expand source code
def implant_patient(filename:str, conn, patient_mode:str=None, same_patient_as:int=None, staff_match:str=None, fallback_staff:str=None, dry_run:bool=False, force:bool=False, progress_callback=None) -> cTransplantResult: """Run a transplant script against the database of <conn>. Settings passed in override the configuration block of the script. Args: conn: connection to the target database, should be gm-dbo patient_mode: 'detect' / 'new' / 'add' same_patient_as: target patient pk to add the data to (implies patient_mode 'add') staff_match: 'strict' / 'lenient' fallback_staff: alias of the staff member to use for unmappable staff dry_run: run everything, then roll back force: allow importing this export again (never duplicates rows) """ if not verify_script_checksum(filename): return cTransplantResult('error', message = _('The transplant script is damaged or has been modified below the configuration block (checksum mismatch).')) settings = {} if patient_mode is not None: settings['patient_mode'] = patient_mode if same_patient_as is not None: settings['same_patient_as'] = str(same_patient_as) settings['patient_mode'] = 'add' if staff_match is not None: settings['staff_match'] = staff_match if fallback_staff is not None: settings['fallback_staff'] = fallback_staff if dry_run: settings['dry_run'] = 'true' if force: settings['force'] = 'true' _log.info('running transplant script [%s], settings: %s', filename, settings) prev_autocommit = conn.autocommit conn.rollback() conn.autocommit = True curs = conn.cursor() stmt_id = None section = None result_row = None in_tx = False try: for stmt_id, section, sql in _iter_script_statements(filename): if progress_callback and section.split(' ')[0] in ('load', 'ref', 'detect', 'staff', 'admin', 'verify', 'preflight'): progress_callback(section) curs.execute(sql) if section == 'begin': in_tx = True for key, val in settings.items(): curs.execute('SELECT set_config(%(key)s, %(val)s, true)', {'key': 'gm_transplant.%s' % key, 'val': val}) elif section == 'result': result_row = curs.fetchone() elif section == 'commit': in_tx = False except Exception as exc: if in_tx: try: curs.execute('ROLLBACK') except Exception: _log.exception('cannot roll back') return _result_from_exception(exc, stmt_id, section) finally: curs.close() try: conn.autocommit = prev_autocommit except Exception: pass if result_row is None: _log.error('transplant script did not report a result') return cTransplantResult('error', message = _('The transplant script did not report a result.')) summary = json.loads(result_row[0]) _log.info('transplant done: target #%s, patient_mode [%s]', summary.get('target_identity'), summary.get('patient_mode')) return cTransplantResult('imported', summary = summary)Run a transplant script against the database of
. Settings passed in override the configuration block of the script.
- Args
- -----=
conn- connection to the target database, should be gm-dbo
patient_mode- 'detect' / 'new' / 'add'
same_patient_as- target patient pk to add the data to (implies patient_mode 'add')
staff_match- 'strict' / 'lenient'
fallback_staff- alias of the staff member to use for unmappable staff
dry_run- run everything, then roll back
force- allow importing this export again (never duplicates rows)
def read_script_config(filename: str) ‑> dict-
Expand source code
def read_script_config(filename:str) -> dict: """Read the (possibly manually edited) configuration block. Best effort, for display purposes only. """ config = {'staff_map': []} in_cfg = False in_staff = False with open(filename, 'rb') as script: for raw_line in script: line = raw_line.decode('utf8', errors = 'replace').rstrip('\n') if line.startswith(MARKER_DATA_BEGIN): break if line.startswith(MARKER_CFG_BEGIN): in_cfg = True continue if line.startswith(MARKER_CFG_END): break if not in_cfg or line.lstrip().startswith('--'): continue match = _REGEX_CFG_LINE.match(line) if match: val = match.group(2) config[match.group(1)] = None if val.upper() == 'NULL' else _unquote_sql(val) continue if 'pg_temp._gm_tp_staff_map' in line: in_staff = True continue if in_staff: values = re.findall(r"NULL|E?'(?:[^'\\]|\\.|'')*'", line) if len(values) == 4: config['staff_map'].append(tuple([ None if v.upper() == 'NULL' else _unquote_sql(v) for v in values ])) if line.strip().endswith(';'): in_staff = False return configRead the (possibly manually edited) configuration block.
Best effort, for display purposes only.
def read_script_metadata(filename: str) ‑> dict-
Expand source code
def read_script_metadata(filename:str) -> dict: """Read the metadata of a transplant script (without running it).""" with open(filename, 'rb') as script: for line in script: if line.startswith(MARKER_DATA_BEGIN.encode('utf8')): break if line.startswith(MARKER_META.encode('utf8')): return json.loads(line[len(MARKER_META):].decode('utf8')) raise cTransplantError(_('[%s] is not a GNUmed transplant script.') % filename)Read the metadata of a transplant script (without running it).
def run_patient_explant_tool()-
Expand source code
def run_patient_explant_tool(): from Gnumed.business import gmPersonSearch print('praxis branch:', gmPraxis.gmCurrentPraxisBranch.from_first_branch()) print('Note: the change history (audit trail) of the patient can only') print(' be exported when logged in as the database owner (gm-dbo).') exit_code = 0 pat = gmPersonSearch.ask_for_patient() while pat: print('patient: #%s' % pat.ID) def_fname = os.path.join ( gmTools.gmPaths().user_work_dir, 'patients', 'gm-transplant-pk%s.sql' ) % pat.ID fname = gmTools.prompted_input(prompt = 'Script file name', default = def_fname) if fname is None: break try: fname = explant_patient ( pk_identity = pat.ID, filename = fname, progress_callback = lambda msg: print(' %s' % msg) ) print('Transplant script:', fname) print('The script contains patient data, treat it as confidential.') except cTransplantError as exc: print('ERROR: %s' % exc) exit_code = -2 pat = gmPersonSearch.ask_for_patient() return exit_code def run_patient_implant_tool(transplant_filename: str = None)-
Expand source code
def run_patient_implant_tool(transplant_filename:str=None): if transplant_filename: if os.access(transplant_filename, os.F_OK): print('transplant file: %s' % transplant_filename) else: print('ERROR: [%s] does not seem to be an accessible file' % transplant_filename) transplant_filename = None if not transplant_filename: transplant_filename = gmTools.prompted_input(prompt = 'Transplant script file name') if not transplant_filename: return 0 transplant_filename = os.path.abspath(os.path.expanduser(transplant_filename)) try: meta = read_script_metadata(transplant_filename) except (cTransplantError, OSError, ValueError) as exc: print('ERROR: %s' % exc) return -2 print('source : %s' % meta.get('source')) print('source patient: #%s' % meta.get('source_pk')) print('exported : %s' % meta.get('exported_when')) print('rows : %s' % sum(meta.get('counts', {}).values())) settings = {} fallback = gmTools.prompted_input(prompt = 'Alias of the staff member to use for unmappable staff (empty: as configured in the script)') if fallback: settings['fallback_staff'] = fallback.strip() conn = gmPG2.get_connection(readonly = False, pooled = False) try: while True: print('dry run ...') result = implant_patient(transplant_filename, conn, dry_run = True, **settings) if result.status == 'already_imported': print('This export has already been imported into: %s' % ', '.join([ '#%s' % pk for pk in result.identities ])) if gmTools.prompted_input(prompt = 'Import again (no rows will be duplicated) ? [y/N]', default = 'n').strip().lower() != 'y': return 0 settings['force'] = True continue if result.status == 'candidates': print('The patient may already exist in this database:') for cand in result.candidates: print(' #%s (%s)%s' % (cand['pk'], ', '.join(cand.get('reasons', [])), ' [deactivated]' if cand.get('deleted') else '')) answer = gmTools.prompted_input(prompt = '[n]ew patient / [a]dd <pk> / [c]ancel', default = 'c') answer = (answer or 'c').strip().lower() if answer == 'n': settings['patient_mode'] = 'new' continue if answer.startswith('a'): pk = answer[1:].strip() if not pk.isdigit(): print('ERROR: add requires the pk of the patient, say: a 4711') continue settings['same_patient_as'] = int(pk) continue return 0 if result.status != 'dry_run_ok': print('ERROR (SQLSTATE %s, statement %s):' % (result.sqlstate, result.statement)) print(result.message) print('Nothing has been changed in the database.') return -2 print('') print(result.format()) print('') if gmTools.prompted_input(prompt = 'Import now ? [y/N]', default = 'n').strip().lower() != 'y': return 0 result = implant_patient(transplant_filename, conn, dry_run = False, **settings) if result.status != 'imported': print('ERROR (SQLSTATE %s, statement %s):' % (result.sqlstate, result.statement)) print(result.message) print('Nothing has been changed in the database.') return -2 print('Patient imported as #%s.' % result.pk_identity) return 0 finally: conn.close() def sql_literal(value) ‑> str-
Expand source code
def sql_literal(value) -> str: """Render a text value as a single-line SQL E'' literal. Line breaks and other control characters are escaped, such that a literal never spans lines in the script. """ if value is None: return 'NULL' return "E'%s'" % str(value).translate(_E_STRING_ESCAPES)Render a text value as a single-line SQL E'' literal.
Line breaks and other control characters are escaped, such that a literal never spans lines in the script.
def verify_export_completeness(conn=None) ‑> list-
Expand source code
def verify_export_completeness(conn=None) -> list: """Check that all tables referencing patient data are handled. Returns a list of problems (strings, no patient data). An empty list means the database layout is fully covered by the transplant specifications. This guards against silently losing data after future schema changes. """ conn, close_conn = _get_conn(conn) curs = conn.cursor() try: return _verify_export_completeness(curs) finally: curs.close() close_conn()Check that all tables referencing patient data are handled.
Returns a list of problems (strings, no patient data). An empty list means the database layout is fully covered by the transplant specifications. This guards against silently losing data after future schema changes.
def verify_script_checksum(filename: str) ‑> bool-
Expand source code
def verify_script_checksum(filename:str) -> bool: """Check the SHA-256 checksum of the protected part of a script.""" sha = hashlib.sha256() hashing = False expected = None with open(filename, 'rb') as script: for line in script: if line.startswith(MARKER_SHA256.encode('utf8')): expected = line[len(MARKER_SHA256):].strip().decode('ascii') break if hashing: sha.update(line) continue if line.startswith(MARKER_DATA_BEGIN.encode('utf8')): hashing = True if expected is None: _log.error('no checksum in [%s]', filename) return False if sha.hexdigest() != expected: _log.error('checksum mismatch in [%s]', filename) return False return TrueCheck the SHA-256 checksum of the protected part of a script.
Classes
class cTransplantError (*args, **kwargs)-
Expand source code
class cTransplantError(Exception): """Raised when a patient cannot be exported or imported. Messages never contain patient data. """ passRaised when a patient cannot be exported or imported.
Messages never contain patient data.
Ancestors
- builtins.Exception
- builtins.BaseException
class cTransplantResult (status: str,
message: str = None,
sqlstate: str = None,
summary: dict = None,
candidates: list = None,
identities: list = None,
statement: str = None)-
Expand source code
class cTransplantResult: """Outcome of running a transplant script. status: 'imported': patient data has been committed into the target database 'dry_run_ok': script ran successfully but was rolled back as requested 'candidates': the patient seems to exist already, see .candidates 'already_imported': this very export has been imported before, see .identities 'error': something went wrong, see .sqlstate/.message """ def __init__(self, status:str, message:str=None, sqlstate:str=None, summary:dict=None, candidates:list=None, identities:list=None, statement:str=None): self.status = status self.message = message self.sqlstate = sqlstate self.summary = summary or {} self.candidates = candidates or [] self.identities = identities or [] self.statement = statement #-------------------------------------------------------- def _get_pk_identity(self) -> int: return self.summary.get('target_identity') pk_identity = property(_get_pk_identity) #-------------------------------------------------------- def _get_staff_mapping(self) -> list: return self.summary.get('staff', []) staff_mapping = property(_get_staff_mapping) #-------------------------------------------------------- def _get_counts(self) -> dict: return self.summary.get('counts', {}) counts = property(_get_counts) #-------------------------------------------------------- def _get_skipped(self) -> list: return self.summary.get('skipped', []) skipped = property(_get_skipped) #-------------------------------------------------------- def __str__(self) -> str: return '<cTransplantResult: status=%s sqlstate=%s target=%s>' % (self.status, self.sqlstate, self.pk_identity) #-------------------------------------------------------- def format(self, eol:str='\n') -> str: """Human readable summary (no patient data beyond primary keys).""" lines = [] summary = self.summary if summary.get('patient_mode'): lines.append(_('Patient mode: %s') % summary['patient_mode']) if summary.get('target_identity'): lines.append(_('Target patient: #%s') % summary['target_identity']) if summary.get('staff'): lines.append(_('Staff mapping:')) for staff in summary['staff']: lines.append(' %s (%s) -> %s [%s]' % ( staff.get('src_alias'), staff.get('src_db_user'), staff.get('target_alias'), _STAFF_HOW_LABELS.get(staff.get('how'), staff.get('how')) )) counts = summary.get('counts', {}) if counts: lines.append(_('Rows:')) for tbl in sorted(counts): if tbl == 'staff' or tbl.startswith('ref:'): continue by_status = ', '.join([ '%s: %s' % (status, counts[tbl][status]) for status in sorted(counts[tbl]) ]) lines.append(' %s: %s' % (tbl, by_status)) refs = [ tbl for tbl in counts if tbl.startswith('ref:') ] if refs: lines.append(_('Reference data:')) for tbl in sorted(refs): by_status = ', '.join([ '%s: %s' % (status, counts[tbl][status]) for status in sorted(counts[tbl]) ]) lines.append(' %s: %s' % (tbl[4:], by_status)) if summary.get('skipped'): lines.append(_('Skipped rows: %s (see transplant protocol)') % len(summary['skipped'])) return eol.join(lines)Outcome of running a transplant script.
status: 'imported': patient data has been committed into the target database 'dry_run_ok': script ran successfully but was rolled back as requested 'candidates': the patient seems to exist already, see .candidates 'already_imported': this very export has been imported before, see .identities 'error': something went wrong, see .sqlstate/.message
Instance variables
prop counts : dict-
Expand source code
def _get_counts(self) -> dict: return self.summary.get('counts', {}) prop pk_identity : int-
Expand source code
def _get_pk_identity(self) -> int: return self.summary.get('target_identity') prop skipped : list-
Expand source code
def _get_skipped(self) -> list: return self.summary.get('skipped', []) prop staff_mapping : list-
Expand source code
def _get_staff_mapping(self) -> list: return self.summary.get('staff', [])
Methods
def format(self, eol: str = '\n') ‑> str-
Expand source code
def format(self, eol:str='\n') -> str: """Human readable summary (no patient data beyond primary keys).""" lines = [] summary = self.summary if summary.get('patient_mode'): lines.append(_('Patient mode: %s') % summary['patient_mode']) if summary.get('target_identity'): lines.append(_('Target patient: #%s') % summary['target_identity']) if summary.get('staff'): lines.append(_('Staff mapping:')) for staff in summary['staff']: lines.append(' %s (%s) -> %s [%s]' % ( staff.get('src_alias'), staff.get('src_db_user'), staff.get('target_alias'), _STAFF_HOW_LABELS.get(staff.get('how'), staff.get('how')) )) counts = summary.get('counts', {}) if counts: lines.append(_('Rows:')) for tbl in sorted(counts): if tbl == 'staff' or tbl.startswith('ref:'): continue by_status = ', '.join([ '%s: %s' % (status, counts[tbl][status]) for status in sorted(counts[tbl]) ]) lines.append(' %s: %s' % (tbl, by_status)) refs = [ tbl for tbl in counts if tbl.startswith('ref:') ] if refs: lines.append(_('Reference data:')) for tbl in sorted(refs): by_status = ', '.join([ '%s: %s' % (status, counts[tbl][status]) for status in sorted(counts[tbl]) ]) lines.append(' %s: %s' % (tbl[4:], by_status)) if summary.get('skipped'): lines.append(_('Skipped rows: %s (see transplant protocol)') % len(summary['skipped'])) return eol.join(lines)Human readable summary (no patient data beyond primary keys).