107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
""" Oracle client """
|
|
|
|
import logging
|
|
import sys
|
|
|
|
import cx_Oracle
|
|
|
|
from mylib.db import DB, 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 occurred 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}"
|