"""
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 `python-oracledb package <https://oracle.github.io/python-oracledb>`_.
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 subclass of it) plus any database parameters, like required credentials
and any database configuration features/options::
app = ConsoleApp()
cred = {'user': 'username', 'password': 'password'}
system = SystemBase('system-id', app, cred)
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:`~ae.db_ora.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)
the :meth:`~OraDb.connect` method is returning an error message string if the connection could not be established.
after a successful connection all the methods provided by :class:`~ae.db_core.DbBase`, like e.g.
:meth:`~ae.db_core.DbBase.update`, are available via the `ora_db` instance::
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
from typing import Any, Union
import oracledb
from ae.db_core import DbBase, SystemBase # type: ignore # SystemBase is an indirect import from ae.sys_core
__version__ = '0.3.9'
[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 Oracle database.
"""
super().__init__(system)
self.conn: oracledb.Connection | None = None
self.last_err_msg = ""
[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 == "":
make_dsn_args = {_k: _v for _k, _v in conn_args.items() if _k.lower() not in ('user', 'password')}
params = oracledb.ConnectParams(**make_dsn_args)
dsn = params.get_connect_string()
elif dsn.count(':') == 1:
host, rest = dsn.split(':', maxsplit=1)
if dsn.count('/@') == 1: # old style format with service id == host:port/@SID
port, service = rest.split('/@', maxsplit=1)
elif dsn.count('/') == 1: # old style format with service name == host:port/service_name
port, service = rest.split('/', maxsplit=1)
else:
self.last_err_msg = f"{dsn=} has unknown or incomplete service id/name format"
return self.last_err_msg
# noinspection PyTypeChecker
params = oracledb.ConnectParams(host=host, port=port, service_name=service) # oracledb converts port str
dsn = params.get_connect_string()
name_space = "CLIENTCONTEXT" # fetch in Oracle with SELECT SYS_CONTEXT(NAMESPACE, "APP") FROM DUAL
app_ctx = [(name_space, "APP", self.console_app.app_name),
(name_space, "LANG", "Python"),
(name_space, "MOD", "ae.db_ora")]
try:
self.conn = oracledb.connect(dsn=dsn, user=user, password=password, appcontext=app_ctx)
# clientversion() is only available in thick mode; thin mode has no Oracle Client
client_ver = oracledb.clientversion() if not oracledb.is_thin_mode() else "thin-mode"
self.console_app.dpo(f"{self}.connect(): connected client version {client_ver}/{oracledb.apilevel}")
except Exception as ex: # pylint: disable=broad-exception-caught
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 = oracledb.DB_TYPE_TIMESTAMP
elif isinstance(value, (int, float)):
ora_type = oracledb.DB_TYPE_NUMBER
else:
ora_type = oracledb.DB_TYPE_VARCHAR
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)