Module Gnumed.business.gmCoding

GNUmed coding systems handling middleware

Functions

def create_data_source(long_name=None, short_name=None, version=None, source=None, language=None)
Expand source code 
def create_data_source(long_name=None, short_name=None, version=None, source=None, language=None):

                args = {
                        'lname': long_name,
                        'sname': short_name,
                        'ver': version,
                        'src': source,
                        'lang': language
                }

                cmd = "SELECT pk FROM ref.data_source WHERE name_long = %(lname)s AND name_short = %(sname)s AND version = %(ver)s"
                rows = gmPG2.run_ro_queries(queries = [{'sql': cmd, 'args': args}])
                if len(rows) > 0:
                        return rows[0]['pk']

                cmd = """
                        INSERT INTO ref.data_source (name_long, name_short, version, source, lang)
                        VALUES (
                                %(lname)s,
                                %(sname)s,
                                %(ver)s,
                                %(src)s,
                                %(lang)s
                        )
                        RETURNING pk
                """
                rows = gmPG2.run_rw_queries(queries = [{'sql': cmd, 'args': args}], return_data = True)

                return rows[0]['pk']
def get_coded_terms(coding_systems=None, languages=None, order_by=None)
Expand source code 
def get_coded_terms(coding_systems=None, languages=None, order_by=None):

        where_snippets = []
        args = {}

        if coding_systems is not None:
                where_snippets.append("((coding_system = ANY(%(sys)s)) OR (coding_system_long = ANY(%(sys)s))")
                args['sys'] = coding_systems

        if languages is not None:
                where_snippets.append('lang = ANY(%(lang)s)')
                args['lang'] = languages

        cmd = 'select * from ref.v_coded_terms'

        if len(where_snippets) > 0:
                cmd += ' WHERE %s' % ' AND '.join(where_snippets)

        if order_by is not None:
                cmd += ' ORDER BY %s' % order_by

        rows = gmPG2.run_ro_queries(queries = [{'sql': cmd, 'args': args}])

        return rows
def get_data_sources(order_by='name_long, lang, version')
Expand source code 
def get_data_sources(order_by='name_long, lang, version'):
        cmd = 'SELECT * FROM ref.data_source ORDER BY %s' % order_by
        rows = gmPG2.run_ro_queries(queries = [{'sql': cmd}])
        return rows
def get_generic_codes(order_by=None)
Expand source code 
def get_generic_codes(order_by=None):
        if order_by is None:
                order_by = 'true'
        else:
                order_by = 'true ORDER BY %s' % order_by

        cmd = _SQL_get_generic_code % order_by
        rows = gmPG2.run_ro_queries(queries = [{'sql': cmd}])
        return [ cGenericCode(row = {'data': r, 'pk_field': 'pk_generic_code'}) for r in rows ]
def get_generic_linked_codes(order_by=None)
Expand source code 
def get_generic_linked_codes(order_by=None):
        if order_by is None:
                order_by = 'true'
        else:
                order_by = 'true ORDER BY %s' % order_by

        cmd = _SQL_get_generic_linked_codes % order_by
        rows = gmPG2.run_ro_queries(queries = [{'sql': cmd}])
        return [ cGenericLinkedCode(row = {'data': r, 'pk_field': 'pk_lnk_code2item'}) for r in rows ]

Classes

class cGenericCode (aPK_obj: int | dict = None, row: dict = None, link_obj=None)
Expand source code 
class cGenericCode(gmBusinessDBObject.cBusinessDBObject):
        """READ ONLY"""
        _cmd_fetch_payload = _SQL_get_generic_code % "pk_generic_code = %s"
        _cmds_store_payload:list = []
        _updatable_fields:list = []

READ ONLY

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 ]

Ancestors

Inherited members

class cGenericLinkedCode (aPK_obj: int | dict = None, row: dict = None, link_obj=None)
Expand source code 
class cGenericLinkedCode(gmBusinessDBObject.cBusinessDBObject):
        """Represents a generic linked code.

        READ ONLY
        """
        _cmd_fetch_payload = _SQL_get_generic_linked_codes % "pk_lnk_code2item = %s"
        _cmds_store_payload:list = []
        _updatable_fields:list = []

Represents a generic linked code.

READ ONLY

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 ]

Ancestors

Inherited members