In-Memory MDM/Lookup System in Python (db_lookup.py)

import collections, pprint, logging, sys, datetime, argparse

import lookup, cx_Oracle

from lookup import normalizer

__version__ = '0.205b-MVP'

__author__ = 'Noelle Milton Vega'

""" Write Me. """

class DatabaseLookup(lookup.Lookup):

""" An concrete Subclass for methods and Database-driven lookup data.

Base class is here: https://www.prismalytics.io/2018/10/07/lookup-py

"""

# ===================================================================================================

# Sub-class initializer. Sub-class specific variables are set here; while non-subclass specific

# variables are set in the Base-class (and accessible here) via a call to BaseClass.__init__(self, ...).

# ===================================================================================================

def __init__(self, connection, db_type='oracle', lookup_tbl_name_prfx=None, synonym_tbl_name=None):

""" Write Me. """

lookup_tbl_name_prfx = normalizer(lookup_tbl_name_prfx,sre=lookup.base_tc_SRE)

synonym_tbl_name = normalizer(synonym_tbl_name,sre=lookup.base_tc_SRE)

lookup.Lookup.__init__(self, connection, lookup_tbl_name_prfx, synonym_tbl_name) # Initialize Base-Class.

event = self._Event('DEBUG',lookup.now(),'Instantiating class: %s' % (self.__class__.__name__), *tuple([None]*4))

self._write_event(event)

assert db_type in lookup.DB_CONSTANTS, '"%s" is not a supported database type.' % db_type

assert isinstance(self._connection, lookup.DB_CONSTANTS[db_type].conn_type), \

'"connection" must be an instance of %s' % (lookup.DB_CONSTANTS[db_type].conn_type)

self.db_type = db_type

event = self._Event('DEBUG',lookup.now(),'Instantiating cursor from DB connection object.', *tuple([None]*4))

self._write_event(event)

try:

self.cursor = self._connection.cursor()

except cx_Oracle.DatabaseError as e:

error, = e.args

print("Oracle-Error-Code:", error.code)

print("Oracle-Error-Message:", error.message)

sys.exit(254)

event = self._Event('DEBUG',lookup.now(),'Class instantiating complete: %s' % (self.__class__.__name__), *tuple([None]*4))

self._write_event(event)

# ===================================================================================================

# Get database Data Dictionary and store in 'self._data_dictionary'. The structure looks like this:

# ===================================================================================================

# defaultdict(<class 'list'>,

# {'tblName1': ['col1Name', 'col2Name', 'col3Name', ...],

# 'tblName2': ['col1Name', 'col2Name', 'col3Name', ...], ...

# 'tblNameN': ['col1Name', 'col2Name', 'col3Name', ...]})

# ===================================================================================================

def _get_data_dictionary(self,):

""" Write Me. """

query = lookup.DB_CONSTANTS[self.db_type].dde_query

params = {'arg1': self._lookup_tbl_name_prfx.upper() + '%', 'arg2':self._synonym_tbl_name.upper()}

# This is the only time in this DatabaseLookup() class that we need to .upper() table names.

try:

event = self._Event('DEBUG',lookup.now(),'Generating data-dictionary from schema.', *tuple([None]*4))

self._write_event(event)

results = self.cursor.execute(query, params).fetchall()

except cx_Oracle.DatabaseError as e:

error, = e.args

print("Oracle-Error-Code:", error.code)

print("Oracle-Error-Message:", error.message)

sys.exit(254)

self._data_dictionary = {tblName.lower(): [clmn.lower() for clmn in colsStr.split(',')] for tblName,colsStr in results}

# ===================================================================================================

# We lower-case the tblNames & colNames as we build self._data_dictionary.

# ===================================================================================================

# cx_Oracle and/or the DDE SQL statement used, won't complain above if the tables queried for

# don't exist It will just return empty list() for them. That is the reasonable behavior, but the

# Lookup/Synonym system is useless if there are no tables, so we must emit an error and

# 'raise ValueError' exception here.

# Logic next: If Synonym-table DOESN'T exist raise an exception. Next, if the Synonym-table DOES exist,

# then it contributes exactly one (qty. 1) entry to the DDE; so we know that no Lookup-tables exist

# if the length of the DDE is equal to one (qty. 1) in this case, so also raise an Exception. I

# found this out when specifying the following -- from either profiler.py and/or populator.py CLI:

# profiler.py|populator.py [...] --lookup-tbl-name-prfx lookup____ <-- When more than one trailing

# underscore is submitted, it still works (e.g. lookup_country gets pulled in). :(

# POST-MVP: Consider removing this enture next section in favor of updating the DDE SQL statement,

# so that doesn't happen. It also doesn't hurt to leave it just as is! =:)

# ===================================================================================================

if (self._synonym_tbl_name not in self.dde) or (len(self.dde) == 1):

self.cursor.close()

raise ValueError('No SYNONYM-table [%s] and/or no LOOKUP-tables for prefix [%s] in DB-Schema accessed.'

% (self._synonym_tbl_name,self._lookup_tbl_name_prfx))

# ===================================================================================================

#pprint.pprint(self.dde) # self.dde is just a @property for self._data_dictionary

#sys.exit(0)

# ===================================================================================================

# IMPORTANT IMPLEMENTATION NOTE:

# ===================================================================================================

# - The case of the table-names and column-names returned by the query that generates the Data

# Dictionary (DDE) will be unchanged (this is performed above, and is in 'self._data_dictionary').

# - However, when that Data Dictionary is subsequently iterated over to import the Lookup-tables

# and the Synonym-table, everything will be LOWER-CASED! So, 'self._lookup_data' will have

# LOWER-CASE Lookup & Synonym-table names as it's dict keys; and the Attribute-names of the

# namedTuples (Row()s) that store each table row, will be LOWER-CASED column-names of that

# particular table. (See example 'self._lookup_data below').

# - Values of the 'lookup_table=' attribute in Synonym-table Rows() (i.e. the namedTuples in

# 'data_synonym' below) will be Lookup-table names in same case as the DDE. As necessary, this

# program Just-In-Time lower()s them for equality-comparison purposes.

# All of this ensures that the user of this class can provide whatever case they wish for TABLE;

# as well as provide defense against changing case of table and column names in the database.

# ===================================================================================================

# {'data_synonym': [Row(synonym_key='boy', lookup_table='LOOKUP_GENDER', lookup_key='male'),

# Row(synonym_key='girl', lookup_table='LOOKUP_GENDER', lookup_key='female'), ...

# Row(synonym_key='pound', lookup_table='LOOKUP_CURRENCY', lookup_key='gbp'),

# Row(synonym_key='russia', lookup_table='LOOKUP_COUNTRY', lookup_key='rus')],

# 'lookup_age_range': [],

# 'lookup_brand': [],

# 'lookup_country': [Row(pk=1, lookup_key='usa', lookup_value='USA', currency_iso='USD'),

# Row(pk=5, lookup_key='fra', lookup_value='FRA', currency_iso='EUR'), ...

# Row(pk=6, lookup_key='ind', lookup_value='IND', currency_iso='INR'),

# Row(pk=7, lookup_key='gbr', lookup_value='GBR', currency_iso='GBP')], ...}

# ===================================================================================================

def load_lookup_tables(self, cleanse=True, rm_orphans=True):

""" Load collection of in-DB LOOKUP-table(s) and SYNONYM-table into in-MEMORY data structure. """

# -----------------------------------------------------------------------------------------

# Whenever this method is called (possibly multiple times within the same object-session),

# we reset/zero-out data-structures that accumulated data before this call.

# -----------------------------------------------------------------------------------------

self._tbl_regexs = {}

self._tbl_metadata = {}

self._lookup_data = {}

self._data_dictionary = {}

self._tbl_shapes = {}

self._events = {'ERROR':[], 'WARN':[], 'INFO':[], 'DEBUG':[]}

self._lookups_hashes = {} # A dict() of dict() for lookup speedup.

# -----------------------------------------------------------------------------------------

# Discover data-dictionary (i.e. DB structure). Stored in 'self._data_dictionary/self.dde'.

# -----------------------------------------------------------------------------------------

self._get_data_dictionary()

# -----------------------------------------------------------------------------------------

# Iterate data-dictionary to dynamically discover LOOKUP & SYNONYM tables/cols. Store it in '_lookup_data'.

# -----------------------------------------------------------------------------------------

for table_name,column_list in sorted(self.dde.items(), key=lambda x: x[0]):

Row = collections.namedtuple('Row', column_list, rename=False)

query = 'SELECT ' + ','.join(column_list) + ' FROM ' + table_name

# NOTE: Above, it's not possible to bind column-list-string w/ cx_Oracle.

try:

self.cursor.execute(query)

except cx_Oracle.DatabaseError as e:

error, = e.args

print("Oracle-Error-Code:", error.code)

print("Oracle-Error-Message:", error.message)

sys.exit(254)

self._lookup_data[table_name] = [x for x in map(Row._make, self.cursor.fetchall())]

lookup.Lookup._load_table_metadata(self) # Populate self._tbl_metadata.

lookup.Lookup._create_tbl_norm_regexs(self) # Generate per-table compiled REGEXs for keys,vals,cols,tbls

# -----------------------------------------------------------------------------------------

# Cleanse Lookup/Synonym-tables and (re)build_lookups_hashes (b/c cleansing alters them).

# -----------------------------------------------------------------------------------------

self._cleanse_and_build_lookups_hashes(self,rm_orphans=rm_orphans) if cleanse else self._build_lookups_hashes(self)

# -----------------------------------------------------------------------------------------

# - Emit diagnostic messages for dde tables loaded and their shapes.

# -----------------------------------------------------------------------------------------

event = self._Event('INFO',lookup.now(),'LOADED Data-Dictionary: [%d tables loaded]' % (len(self.dde)), *tuple([None]*4))

self._write_event(event)

for tblName,rowList in sorted(self.lookup_data.items()):

numRows,numCols = (len(rowList),len(rowList[0])) if rowList else (0,0)

self._tbl_shapes[tblName] = (numRows,numCols)

event = self._Event('INFO',lookup.now(),'LOADED Table: %s [%d-Rows x %d-Cols]' % (tblName,numRows,numCols),*tuple([None]*4))

self._write_event(event)

# -----------------------------------------------------------------------------------------

#pprint.pprint(self.lookup_data) # self.lookup_data is just a @property for self._lookup_data

#pprint.pprint(self.events) # self.events is just a @property for self._events

#sys.exit(0)

return (True, self.lookup_data)

# ===================================================================================================

# Alias for load_lookup_tables() that semantically denotes RE-load (as opposed to a first-time load).

# ===================================================================================================

reload_lookup_tables = load_lookup_tables

# ===================================================================================================

# Lorem Ipsum ...

# ===================================================================================================

def lookup_key(self, lookup_tbl_name, record_key):

""" Write Me. """

synonym_tbl_name = self._synonym_tbl_name # Already normalizer()ed. self._lookup_tbl_name_prfx, too.

lookup_tbl_name = self._lookup_tbl_name_prfx + normalizer(lookup_tbl_name,sre=lookup.base_tc_SRE)

record_key_Lnorm = normalizer(record_key,sre=self.regexs[lookup_tbl_name].key) # Normalized for Lookup-table lookups.

record_key_Snorm = normalizer(record_key,sre=self.regexs[synonym_tbl_name].key) # Normalized for Synonym-table lookups.

# -----------------------------------------------------------------------------------------

# If specified LOOKUP or SYNONYM table DOES NOT EXIST, raise exception. Means feeddef incorrect.

# -----------------------------------------------------------------------------------------

if lookup_tbl_name not in self.lookup_data: # LOOKUP-table doesn't exist.

raise KeyError('LOOKUP_TABLE_UNKNOWN: {}'.format(lookup_tbl_name))

if synonym_tbl_name not in self.lookup_data: # SYNONYM-table doesn't exist.

raise KeyError('SYNONYM_TABLE_UNKNOWN: {}'.format(synonym_tbl_name))

# -----------------------------------------------------------------------------------------

# STEP-1: Search for record_key in the LOOKUP-table (if it's not empty).

# Example Row (for LOOKUP_CURRENCY): Row(PK=23, LOOKUP_KEY='usa', LOOKUP_VALUE='USA', CURRENCY_ISO='USD')

# -----------------------------------------------------------------------------------------

if not self.lookup_data[lookup_tbl_name]: # LOOKUP-table exists, but is empty.

event = self._Event('DEBUG',lookup.now(),'LOOKUP TABLE EMPTY: %s' %(lookup_tbl_name),*tuple([None]*4))

self._write_event(event)

return (None,None) # LOOKUP-table is empty, so 'return(None,None)'.

if record_key_Lnorm in self._lookups_hashes[lookup_tbl_name]:

L_indx = self._lookups_hashes[lookup_tbl_name][record_key_Lnorm]

L_row = self.lookup_data[lookup_tbl_name][L_indx]

event = self._Event('DEBUG',lookup.now(),'Found requested record in LOOKUP-table: %s[%s]'

%(lookup_tbl_name,str(L_row)),*tuple([None]*4))

self._write_event(event)

return (L_row[0],L_row) # Found it and are returning it.

event = self._Event('DEBUG',lookup.now(),'Normalized-KEY [%s] not in LOOKUP-table:[%s]. Searching SYNONYM-table:[%s]'

%(record_key,lookup_tbl_name,synonym_tbl_name),*tuple([None]*4))

self._write_event(event)

# -----------------------------------------------------------------------------------------

# STEP-2: If record_key is NOT found in LOOKUP-table (above), we check the SYNONYM-table.

# Example Row (DATA_SYNONYM): Row(SYNONYM_KEY='USofA', LOOKUP_TABLE='LOOKUP_CURRENCY', LOOKUP_KEY='usa')

# -----------------------------------------------------------------------------------------

# STEP-2A: First check if the SYNONYM-table does/doesn't exist.

# -----------------------------------------------------------------------------------------

if not self.lookup_data[synonym_tbl_name]: # SYNONYM-table exists, but is empty.

event = self._Event('DEBUG',lookup.now(),'SYNONYM TABLE EMPTY: %s' %(synonym_tbl_name),*tuple([None]*4))

self._write_event(event)

event = self._Event('DEBUG',lookup.now(),'Normalized-KEY [%s] not in LOOKUP-table [%s] nor in empty SYNONYM-table [%s].'

%(record_key,lookup_tbl_name,synonym_tbl_name),*tuple([None]*4))

self._write_event(event)

return (None,None) # record_key not in LOOKUP-table; and SYNONYM-table is empty, so 'return(None,None)'.

# -----------------------------------------------------------------------------------------

# STEP-2B: SYNONYM-table exists, so search it and if we get a hit, return it's Lookup-table record.

# -----------------------------------------------------------------------------------------

if (record_key_Snorm + ':' + lookup_tbl_name) in self._lookups_hashes[synonym_tbl_name]:

S_indx = self._lookups_hashes[synonym_tbl_name][record_key_Snorm + ':' + lookup_tbl_name]

S_row = self.lookup_data[synonym_tbl_name][S_indx]

lookback_key = normalizer(S_row[2],sre=self.regexs[lookup_tbl_name].key)

if lookback_key in self._lookups_hashes[lookup_tbl_name]:

L_indx = self._lookups_hashes[lookup_tbl_name][lookback_key]

L_row = self.lookup_data[lookup_tbl_name][L_indx]

return (L_row[0],L_row) # Found it and are returning it.

# -----------------------------------------------------------------------------------------

event = self._Event('DEBUG',lookup.now(),'Normalized-KEY [%s] not in LOOKUP-table [%s] nor SYNONYM-table [%s].'

%(record_key,lookup_tbl_name,synonym_tbl_name),*tuple([None]*4))

self._write_event(event)

# -----------------------------------------------------------------------------------------

return (None, None)

# ===================================================================================================

# Lorem Ipsum ...

# IMPORTANT: As far as the 'lookup_tbl_name' and 'record_key' arguments passed to

# this lookup_attribute() function go, DO NOT str(), strip(), lower(), upper(),

# normalizer() or tramsform anything about them here (DO NOT!). REASON: This section

# calls/wraps the lookup_key() function, which peforms whatever transformations necessary.

# ===================================================================================================

def lookup_attribute(self, lookup_tbl_name, record_key, attribute_col_name):

""" Write Me. """

(pk,row) = self.lookup_key(lookup_tbl_name, record_key)

attribute_col_name = normalizer(attribute_col_name,sre=lookup.base_tc_SRE)

if (pk,row) == (None, None): return (None, None) # lookup_key() may not find a match.

if pk and attribute_col_name not in row._fields:

raise AttributeError('LOOKUP_UNKNOWN_COLUMN: {}'.format(attribute_col_name))

return (row._asdict()[attribute_col_name], row)

# ===================================================================================================

# This class should be instantiated using Python's with/as construct, as shown here. Doing so enables

# the class to exetue setup code (via __enter__) and tear-down code (via __exit__).

#

# >>> import lookup

# >>> with lookup.DatabaseLookup(connection=c, db_type=t) as lkup:

# # lkup.load_lookup_tables()

# # Do ther stuff with the 'lkup' object.

# >>>

# ===================================================================================================

def __enter__(self,):

return self

def __exit__(self, exc_type, exc_value, traceback):

try:

event = self._Event('DEBUG',lookup.now(),'Closing cursor from DB connection object.',*tuple([None]*4))

self._write_event(event)

self.cursor.close()

#event = self._Event('DEBUG',lookup.now(),'Closing DB connection object.',*tuple([None]*4))

#self._write_event(event)

#self._connection.close()

except cx_Oracle.DatabaseError as e:

error, = e.args

print("Oracle-Error-Code:", error.code)

print("Oracle-Error-Message:", error.message)

sys.exit(254)

# ===================================================================================================

if __name__ == '__main__':

logging.basicConfig(level=logging.ERROR)

api_version = lookup.__version__

DB_TYPE = 'oracle' # Hardcode this to jusr Oracle for MVP.

DFLT_LOOKUP_TBL_NAME_PRFX = lookup.Lookup.DFLT_LOOKUP_TBL_NAME_PRFX

DFLT_SYNONYM_TBL_NAME = lookup.Lookup.DFLT_SYNONYM_TBL_NAME

DFLT_DB_CONN_STRING = 'scott/tiger@127.0.0.1:1521/oracle'

def parse_cli():

# -----------------------------------------------------------------------------------------------

parser = argparse.ArgumentParser(add_help=True, description=None,

formatter_class=argparse.RawDescriptionHelpFormatter)

required = parser.add_argument_group('required args')

# -----------------------------------------------------------------------------------------------

parser.add_argument('--version', action='version', version='%(prog)s ' + __version__)

required.add_argument('--lookup-tbl-name-prfx', action='store', dest='lookup_tbl_name_prfx',

type=str, nargs=1, default=[DFLT_LOOKUP_TBL_NAME_PRFX,], required=False,

help='LOOKUP-table name prefix. (Default: "%s")' % DFLT_LOOKUP_TBL_NAME_PRFX)

required.add_argument('--synonym-tbl-name', action='store', dest='synonym_tbl_name',

type=str, nargs=1, default=[DFLT_SYNONYM_TBL_NAME,], required=False,

help='SYNONYM-table name. (Default: "%s")' % DFLT_SYNONYM_TBL_NAME)

required.add_argument('--db-conn-string', action='store', dest='db_conn_string', type=str,

nargs=1, default=[DFLT_DB_CONN_STRING,], required=False,

help='Database connection string. (Default: "%s")' % DFLT_DB_CONN_STRING)

required.add_argument('--db-type', action='store', dest='db_type', type=str,

nargs=1, default=[DB_TYPE,], required=False,

help='Database type. (Default: "%s")' % DB_TYPE)

# -----------------------------------------------------------------------------------------------

return parser.parse_args()

# ===============================================================================================

# Parse the command-line/CLI, and return dictionary (cli_dict) that looks like this:

# {'files': ['aFile','bFile','cFile'], 'columns': ['aCol','bCol','cCol'], 'table': ['aTable' |'NIL_TABLE']}

# ===============================================================================================

cli_dict = vars(parse_cli())

# ===============================================================================================

lookup_tbl_name_prfx = cli_dict['lookup_tbl_name_prfx'][0]

lookup_tbl_name_prfx = normalizer(lookup_tbl_name_prfx,sre=lookup.base_tc_SRE)

synonym_tbl_name = cli_dict['synonym_tbl_name'][0]

synonym_tbl_name = normalizer(synonym_tbl_name, sre=lookup.base_tc_SRE)

db_conn_string = cli_dict['db_conn_string'][0]

db_type = cli_dict['db_type'][0]

# ===============================================================================================

connection = cx_Oracle.connect(db_conn_string)

L = DatabaseLookup(connection=connection, db_type=db_type,

lookup_tbl_name_prfx=lookup_tbl_name_prfx, synonym_tbl_name=synonym_tbl_name)

L.load_lookup_tables(cleanse=True, rm_orphans=True)

# ===============================================================================================

# Class test code goes in between here.

# ===============================================================================================

#pprint.pprint(L.lookup_data['lookup_country']) # Tests __getitem__()

#pprint.pprint(L.lookup_data['data_synonym']) # Tests __getitem__()

#for item in L: pprint.pprint(item) # Tests __iter__()

#pprint.pprint(L.dde)

#pprint.pprint(L.events)

#pprint.pprint(L._lookups_hashes['lookup_country'])

#pprint.pprint(L._lookups_hashes['data_synonym'])

#pprint.pprint(L._lookups_hashes.keys())

# ===============================================================================================

# Class test code goes in between here.

# ===============================================================================================

L.cursor.close()

connection.close()