Untitled
unknown
plain_text
2 years ago
1.9 kB
2
Indexable
import mysql.connector import sqlalchemy from sqlalchemy import create_engine from base64 import b64decode def connect_to_database(json_data, replicationTarget): try: encrypt = json_data.get(replicationTarget, {}).get('ENCRYPT') host = json_data.get(replicationTarget, {}).get('DB_HOST') port = json_data.get(replicationTarget, {}).get('DB_PORT') user = json_data.get(replicationTarget, {}).get('DB_USER') db_type = json_data.get(replicationTarget, {}).get('DB_TYPE') schema = json_data.get(replicationTarget, {}).get('DB_SCHEMA') if encrypt == 'Y': password = b64decode(json_data.get(replicationTarget, {}).get('DB_PASSWORD')).decode('utf-8') else: password = json_data.get(replicationTarget, {}).get('DB_PASSWORD') if db_type == 'MYSQL': cnx = mysql.connector.connect(user=user, password=password, host=host, port=port) cursor = cnx.cursor() logging.info(f"Connected to MySQL database server {replicationTarget}: {host}:{port}") elif db_type == 'ORACLE': import oracledb oracle_mode = oracledb.is_thin_mode() print("Oracle mode: %s" % oracle_mode) if oracle_mode: oracledb.init_oracle_client() print("Enabled python-oracledb Thick mode") else: print("Default python-oracledb Thick mode") cnx_text = ('oracle://%s:%s@%s:%s/?service_name=%s' % (user, password, host, port, schema)) cnx = create_engine(cnx_text, encoding="utf8").raw_connection() cursor = cnx.cursor() return cnx, cursor, schema except (mysql.connector.Error, sqlalchemy.exc.SQLAlchemyError, ImportError, Exception) as e: logging.error(f"An error occurred while connecting to the database: {str(e)}") raise e
Editor is loading...