import oracledb as db def copy_data(self): src = self.src_db.get_connection(self.src_dsn) tgt = self.tgt_db.get_connection(self.tgt_dsn) c1 = src.cursor() c2 = tgt.cursor() rows_affected = 0 commit_rate = 5000 update_cycle = 50000 if self.o_src_table.column_list != self.o_tgt_table.column_list: log.warning("Structure between Source and Target Table does not match.") try: c1.execute(self.src_query) if self.src_type == self.tgt_type and self.src_type == 'ORACLE': db_types = (x[1] for x in c1.description) c2.setinputsizes(*db_types) [log.info](http://log.info/)("Copy data with blocksize: {0}".format(commit_rate)) i_update = update_cycle while True: data = c1.fetchmany(commit_rate) if len(data) == 0: break c2.executemany(self.tgt_query, data) rows_affected += c2.rowcount if rows_affected > i_update: meta_db.update_repo(self.dsn, self.app_name, self.job_name, self.successful_rows, rows_affected) i_update += update_cycle tgt.commit() if data: del data gc.collect() meta_db.update_repo(self.dsn, self.app_name, self.job_name, self.successful_rows, rows_affected) except self.src_db.db.Error as err: log.error("Source DB-Error-Code: {0}".format(err)) return False except self.tgt_db.db.Error as err: log.error("Target DB-Error-Code: {0}".format(err)) return False finally: c1.close() c2.close() src.close() tgt.close() return rows_affected def output_typehandler(cursor, name, defaultType, size, precision, scale): """ Set workaround for CLOB / BLOB data types on connection level. """ if defaultType == db.DB_TYPE_CLOB: return cursor.var(db.DB_TYPE_LONG, arraysize=cursor.arraysize) elif defaultType == db.DB_TYPE_BLOB: return cursor.var(db.DB_TYPE_LONG_RAW, arraysize=cursor.arraysize) def get_connection(connection_key): cred = seclib.get_credentials(connection_key) if cred is False: return False try: con = db.connect(user="{0}".format(cred[1]), password="{0}".format(cred[2]), dsn="{0}".format(cred[0])) con.clientinfo = "H.A.S.I. on {0}".format(host) con.module = "oralib.py" con.action = "SQL Query" log.debug("Connected to {0}, version {1} successful.".format(con.dsn, con.version)) con.outputtypehandler = output_typehandler except db.Error as exc: err, = exc.args log.error("Oracle-Error-Code: {0}".format(err.code)) log.error("Oracle-Error-Message: {0}".format(err.message)) log.error("Key: {0}".format(connection_key)) return False return con def create_selector(l_intersect, o_src_table): """ Generate SQL statement for selector """ l_selection = [] """ XML workaround Oracle """ xml_cols = list(x for x in l_intersect if o_src_table.d_col_data_type[x][0] == 'XMLTYPE') for x in l_intersect: if x in xml_cols: l_selection.append("xmltype.getclobval({0}.\"{1}\".\"{2}\") as {2}".format( o_src_table.schema, o_src_table.table_name, x)) elif x not in xml_cols: l_selection.append("\"{0}\".\"{1}\".\"{2}\"".format(o_src_table.schema, o_src_table.table_name, x)) return "SELECT {0} FROM \"{1}\".\"{2}\"".format(", ".join(l_selection), o_src_table.schema, o_src_table.table_name) def create_dml(l_intersect, o_src_table, o_tgt_table, src_filter, action): """ Generate SQL statement for dml """ l_placeholder = list(":\"{0}\"".format(x) for x in l_intersect) """ XML workaround Oracle """ xml_cols = list(x for x in l_intersect if o_tgt_table.d_col_data_type[x][0] == 'XMLTYPE') for ix, col in enumerate(l_intersect): if col in xml_cols: l_placeholder[ix] = "xmltype.createxml(:\"{0}\")".format(col) if action in ("INSERT", "MASK"): return "INSERT INTO \"{0}\".\"{1}\" (\"{2}\") VALUES ({3})".format( o_tgt_table.schema, o_tgt_table.table_name, "\", \"".join(l_intersect), ", ".join(l_placeholder)) elif action in ("UPSERT", "UPSERT_MASK"): return "UPSERT INTO \"{0}\".\"{1}\" (\"{2}\") VALUES ({3})".format( o_tgt_table.schema, o_tgt_table.table_name, "\", \"".join(l_intersect), ", ".join(l_placeholder)) elif action == "MERGE": exp_join = " and ".join(["src." + x + "=tgt." + x for x in o_tgt_table.primary_key_list]) exp_merge_update = ", ".join(["tgt." + x + "=src." + x for x in o_tgt_table.column_list]) exp_merge_tgt_rows = ", ".join(["tgt." + x for x in o_tgt_table.column_list]) if len(o_tgt_table.column_list) - len(o_src_table.column_list) == 1 \ and o_tgt_table.column_list[-1] == "TA_FEHLER": exp_merge_src_rows = ", ".join(["src." + x for x in o_tgt_table.column_list]) tgt_query = "merge into {0}.{1} tgt using " \ "(select {2}, 'Primärschlüssel unvollständig' as TA_FEHLER " \ "from {3}.{4} {5}) src on ({6}) " \ "when matched then update set {7} " \ "when not matched then insert ({8}) values ({9});".format( o_tgt_table.schema, o_tgt_table.table_name, ", ".join(o_src_table.column_list), o_src_table.schema, o_src_table.table_name, src_filter, exp_join, exp_merge_update, exp_merge_tgt_rows, exp_merge_src_rows) o_src_table.l_src_cols.append("'Primärschlüssel unvollständig.' as TA_FEHLER") else: exp_merge_src_rows = ", ".join(["src." + x for x in o_src_table.l_src_cols]) tgt_query = "merge into {0}.{1} tgt using " \ "(select {2} from {3}.{4} {5}) src on ({6}) " \ "when matched then update set {7} " \ "when not matched then insert ({8}) values ({9});".format( o_tgt_table.schema, o_tgt_table.table_name, ", ".join(o_src_table.column_list), o_src_table.schema, o_src_table.table_name, src_filter, exp_join, exp_merge_update, exp_merge_tgt_rows, exp_merge_src_rows) return tgt_query