"""
database system core layer to access Oracle databases
=====================================================
the class :class:`OraDb` of this namespace portion is a thin layer that is extending the :class:`~ae.db_core.DbBase` of
the module :mod:`ae.db_core` to connect to an Oracle database.
.. hint::
this namespace portion is using the `cx_Oracle package <https://cx-oracle.readthedocs.io/en/7.2/module.html>`_
as the database driver. the cx_Oracle package has to have at least version 5 or higher.
basic usage of the oracle database layer
----------------------------------------
to create an instance of the class :class:`OraDb` you first have to create a :class:`~ae.sys_core.SystemBase` instance.
this can be done either programmatically by providing an application instance (of the class
:class:`~ae.console.ConsoleApp` or an inherited sub-class of it) plus any database parameters, like required credentials
and any database configuration features/options::
app = ConsoleApp()
system = SystemBase('system-id', app, dict(User='user name', Password='password`, ...), ...)
alternatively provide all system-specific info within the :ref:`ae config files<config-files>` and let
:class:`~ae.sys_core.UsedSystems` load it::
system = used_systems['system-id']
finally pass the database parameters in `system` to create an instance of :class:`OraDb`::
ora_db = OraDb(system)
then call the :meth:`~OraDb.connect` method of this instance to connect to the Oracle database server::
error_message = ora_db.connect()
if error_message:
print(error_message)
if the connection could not be established then :meth:`~OraDb.connect` is returning an error message string. if the
return value is an empty string then you can use all the methods provided by :class:`~ae.db_core.DbBase`, like e.g.
:meth:`~ae.db_core.DbBase.update`::
error_message = ora_db.update('my_table`, {'my_col': 'new value'})
if error_message:
print(error_message)
error_message = ora_db.rollback()
an explicit call of :meth:`~ae.db_core.DbBase.rollback` is only needed if you use transactions. in this case you should
also use :meth:`~ae.db_core.DbBase.commit` at the end of each transaction to store any data updates::
error_message = ora_db.commit()
alternatively you can use the `commit` argument that is provided by the :class:`~ae.db_core.DbBase` DML methods: by
passing a `True` value to this argument, the method will automatically execute a :meth:`~ae.db_core.DbBase.commit` call
for you if no error occurred in the DML method::
error_message = ora_db.update('table`, {'column': 369}, commit=True)
finally after all database actions are done you can close the connection to the databases server with the
:meth:`~ae.db_core.DbBase.close` method::
error_message = ora_db.close()
"""
import datetime
import os
from typing import Any, Union
import cx_Oracle # type: ignore
from ae.db_core import DbBase, SystemBase # type: ignore # SystemBase is an indirect import from ae.sys_core
__version__ = '0.3.7'
[docs]class OraDb(DbBase):
""" Oracle database class, based on :class:`~.db_core.DbBase` """
[docs] def __init__(self, system: SystemBase):
""" create instance of oracle database object.
:param system: instance of a :class:`~ae.sys_core.SystemBase` class.
:class:`~ae.sys_core.SystemBase` (defined in the module :mod:`ae.sys_core`) is providing
the credentials and features, which get retrieved from :ref:`config-files`, then converted
by :meth:`~ae.db_core.DbBase.connect_params` into :ref:`connection parameters` to connect
to the Postgres database.
if you experiencing the following unicode encoding error::
'charmap' codec can't decode byte 0x90 in position 2: character maps to <undefined>
don't try to create a type handler like recommended in some places - still got same error after adding
the following method to replace the self.conn.outputtypehandler of the database driver::
def output_type_handler(cursor, name, default_type, size, precision, scale):
if default_type in (cx_Oracle.STRING, cx_Oracle.FIXED_CHAR):
return cursor.var(cx_Oracle.NCHAR, size, cursor.arraysize)
luckily, finally found workaround by setting the following OS environment variable to the
character set of the used Oracle server (here UTF8)::
os.environ["NLS_LANG"] = ".AL32UTF8"
"""
super().__init__(system)
os.environ["NLS_LANG"] = '.AL32UTF8'
[docs] def connect(self) -> str:
""" connect this instance to the database driver. """
self.last_err_msg = ''
conn_args = self.connect_params()
user = conn_args.get('user')
password = conn_args.get('password')
dsn: str = conn_args.get('dsn')
if dsn:
if dsn.count(':') == 1 and dsn.count('/@') == 1: # old style format == host:port/@SID
host, rest = dsn.split(':', maxsplit=1)
port, service_id = rest.split('/@', maxsplit=1)
dsn = cx_Oracle.makedsn(host=host, port=port, sid=service_id)
elif dsn and dsn.count(':') == 1 and dsn.count('/') == 1: # old style format == host:port/service_name
host, rest = dsn.split(':', maxsplit=1)
port, service_name = rest.split('/', maxsplit=1)
dsn = cx_Oracle.makedsn(host=host, port=port, service_name=service_name)
else:
make_dsn_args = {k: v for k, v in conn_args.items() if k not in ('user', 'password')}
dsn = cx_Oracle.makedsn(**make_dsn_args)
app_name = self.console_app.app_name
try:
# connect old style (using conn str): cx_Oracle.connect(self.usr + '/"' + self.pwd + '"@' + self.dsn)
if cx_Oracle.__version__ >= '6':
# sys context is using appcontext kwarg starting with cx_Oracle Version 6, which is
# .. list of 3-tuples. so since V6 need to replace clientinfo kwarg with appcontext=app_ctx
name_space = "CLIENTCONTEXT" # fetch in Oracle with SELECT SYS_CONTEXT(NAMESPACE, "APP") FROM DUAL
app_ctx = [(name_space, "APP", app_name),
(name_space, "LANG", "Python"),
(name_space, "MOD", "ae.db_ora")]
self.conn = cx_Oracle.connect(user=user, password=password, dsn=dsn, appcontext=app_ctx)
else:
# sys context old style (until cx_Oracle Version 5 using clientinfo):
self.conn = cx_Oracle.connect(user=user, password=password, dsn=dsn, clientinfo=app_name)
# self.conn.outputtypehandler = output_type_handler # see also comment in OraDb.__init__()
self.console_app.dpo(f"OraDb.connect(): connected"
f" via client version {cx_Oracle.clientversion()}/{cx_Oracle.apilevel}"
f" with n-/encoding {self.conn.nencoding}/{self.conn.encoding} for {self}")
except Exception as ex:
self.last_err_msg = f"OraDb-connect() error '{ex}' for {self}"
else:
self.create_cursor()
return self.last_err_msg
[docs] def prepare_ref_param(self, value: Union[datetime.datetime, int, float, str]) -> Any:
""" prepare special Oracle reference parameter.
:param value: the input value passed into the reference parameter of the called stored procedure.
:return: a handle to the reference variable.
the following code snippet shows how to use this method together with :meth:`~.get_value` to retrieve the
returned value of a reference parameter::
ora_db = OraDb(...)
*ref_var* = ora_db.prepare_ref_param("input_value")
err_msg = ora_db.call_proc('STORED_PROCEDURE', (*ref_var*, ...))
if not err_msg:
output_value = ora_db.get_value(*ref_var*)
"""
if isinstance(value, datetime.datetime): # also True if value is datetime.date because inherits from datetime
ora_type = cx_Oracle.DATETIME
elif isinstance(value, int) or isinstance(value, float):
ora_type = cx_Oracle.NUMBER
else:
ora_type = cx_Oracle.STRING
value = str(value)
ref_var = self.curs.var(ora_type)
if value is not None:
self.set_value(ref_var, value)
return ref_var
[docs] @staticmethod
def get_value(var) -> Any:
""" get output value from a reference variable passed into a stored procedure.
:param var: handle to a reference variable.
:return: output value of the reference variable.
"""
return var.getvalue()
[docs] @staticmethod
def set_value(var: Any, value: Union[datetime.datetime, int, float, str]):
""" set the input value of a reference variable to pass into a stored procedure.
:param var: handle to the reference variable to set.
:param value: value to set as input value of the reference variable.
"""
var.setvalue(0, value)