python-mylib/mylib/oracle.py

# -*- 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}'