# -*- coding: utf-8 -*- """ Oracle client """ import logging import sys import cx_Oracle from mylib.db import DB from mylib.db import DBFailToConnect log = logging.getLogger(__name__) class OracleDB(DB): """ Oracle client """ _dsn = None _user = None _pwd = None def __init__(self, dsn, user, pwd, **kwargs): self._dsn = dsn self._user = user self._pwd = pwd super().__init__(**kwargs) def connect(self, exit_on_error=True): """ Connect to Oracle server """ if self._conn is None: log.info('Connect on Oracle server with DSN %s as %s', self._dsn, self._user) try: self._conn = cx_Oracle.connect( user=self._user, password=self._pwd, dsn=self._dsn ) except cx_Oracle.Error as err: log.fatal( 'An error occured during Oracle database connection (%s@%s).', self._user, self._dsn, exc_info=1 ) if exit_on_error: sys.exit(1) else: raise DBFailToConnect(f'{self._user}@{self._dsn}') from err return True def doSQL(self, sql, params=None): """ Run SQL query and commit changes (rollback on error) :param sql: The SQL query :param params: The SQL query's parameters as dict (optional) :return: True on success, False otherwise :rtype: bool """ if self.just_try: log.debug("Just-try mode : do not really execute SQL query '%s'", sql) return True try: self._log_query(sql, params) with self._conn.cursor() as cursor: if isinstance(params, dict): cursor.execute(sql, **params) else: cursor.execute(sql) self._conn.commit() return True except cx_Oracle.Error: self._log_query_exception(sql, params) self._conn.rollback() return False def doSelect(self, sql, params=None): """ Run SELECT SQL query and return list of selected rows as dict :param sql: The SQL query :param params: The SQL query's parameters as dict (optional) :return: List of selected rows as dict on success, False otherwise :rtype: list, bool """ try: self._log_query(sql, params) with self._conn.cursor() as cursor: if isinstance(params, dict): cursor.execute(sql, **params) else: cursor.execute(sql) cursor.rowfactory = lambda *args: dict( zip([d[0] for d in cursor.description], args) ) results = cursor.fetchall() return results except cx_Oracle.Error: self._log_query_exception(sql, params) return False # # SQL helpers # @staticmethod def format_param(param): """ Format SQL query parameter for prepared query """ return f':{param}'