in providers/oracle/src/airflow/providers/oracle/hooks/oracle.py [0:0]
def get_conn(self) -> oracledb.Connection:
"""
Get an Oracle connection object.
Optional parameters for using a custom DSN connection (instead of using
a server alias from tnsnames.ora) The dsn (data source name) is the TNS
entry (from the Oracle names server or tnsnames.ora file), or is a
string like the one returned from ``makedsn()``.
:param dsn: the data source name for the Oracle server
:param service_name: the db_unique_name of the database
that you are connecting to (CONNECT_DATA part of TNS)
:param sid: Oracle System ID that identifies a particular
database on a system
You can set these parameters in the extra fields of your connection
as in
.. code-block:: python
{"dsn": ("(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=host)(PORT=1521))(CONNECT_DATA=(SID=sid)))")}
see more param detail in `oracledb.connect
<https://python-oracledb.readthedocs.io/en/latest/api_manual/module.html#oracledb.connect>`_
"""
conn = self.get_connection(self.oracle_conn_id) # type: ignore[attr-defined]
conn_config = {"user": conn.login, "password": conn.password}
sid = conn.extra_dejson.get("sid")
mod = conn.extra_dejson.get("module")
schema = conn.schema
# Enable oracledb thick mode if thick_mode is set to True
# Parameters take precedence over connection config extra
# Defaults to use thin mode if not provided in params or connection config extra
thick_mode = _get_first_bool(self.thick_mode, conn.extra_dejson.get("thick_mode"))
if thick_mode is True:
if self.thick_mode_lib_dir is None:
self.thick_mode_lib_dir = conn.extra_dejson.get("thick_mode_lib_dir")
if not isinstance(self.thick_mode_lib_dir, (str, type(None))):
raise TypeError(
f"thick_mode_lib_dir expected str or None, "
f"got {type(self.thick_mode_lib_dir).__name__}"
)
if self.thick_mode_config_dir is None:
self.thick_mode_config_dir = conn.extra_dejson.get("thick_mode_config_dir")
if not isinstance(self.thick_mode_config_dir, (str, type(None))):
raise TypeError(
f"thick_mode_config_dir expected str or None, "
f"got {type(self.thick_mode_config_dir).__name__}"
)
oracledb.init_oracle_client(
lib_dir=self.thick_mode_lib_dir, config_dir=self.thick_mode_config_dir
)
# Set oracledb Defaults Attributes if provided
# (https://python-oracledb.readthedocs.io/en/latest/api_manual/defaults.html)
fetch_decimals = _get_first_bool(self.fetch_decimals, conn.extra_dejson.get("fetch_decimals"))
if isinstance(fetch_decimals, bool):
oracledb.defaults.fetch_decimals = fetch_decimals
fetch_lobs = _get_first_bool(self.fetch_lobs, conn.extra_dejson.get("fetch_lobs"))
if isinstance(fetch_lobs, bool):
oracledb.defaults.fetch_lobs = fetch_lobs
# Set up DSN
service_name = conn.extra_dejson.get("service_name")
port = conn.port if conn.port else DEFAULT_DB_PORT
if conn.host and sid and not service_name:
conn_config["dsn"] = oracledb.makedsn(conn.host, port, sid)
elif conn.host and service_name and not sid:
conn_config["dsn"] = oracledb.makedsn(conn.host, port, service_name=service_name)
else:
dsn = conn.extra_dejson.get("dsn")
if dsn is None:
dsn = conn.host
if conn.port is not None:
dsn += f":{conn.port}"
if service_name:
dsn += f"/{service_name}"
conn_config["dsn"] = dsn
if "events" in conn.extra_dejson:
conn_config["events"] = conn.extra_dejson.get("events")
mode = conn.extra_dejson.get("mode", "").lower()
if mode == "sysdba":
conn_config["mode"] = oracledb.AUTH_MODE_SYSDBA
elif mode == "sysasm":
conn_config["mode"] = oracledb.AUTH_MODE_SYSASM
elif mode == "sysoper":
conn_config["mode"] = oracledb.AUTH_MODE_SYSOPER
elif mode == "sysbkp":
conn_config["mode"] = oracledb.AUTH_MODE_SYSBKP
elif mode == "sysdgd":
conn_config["mode"] = oracledb.AUTH_MODE_SYSDGD
elif mode == "syskmt":
conn_config["mode"] = oracledb.AUTH_MODE_SYSKMT
elif mode == "sysrac":
conn_config["mode"] = oracledb.AUTH_MODE_SYSRAC
purity = conn.extra_dejson.get("purity", "").lower()
if purity == "new":
conn_config["purity"] = oracledb.PURITY_NEW
elif purity == "self":
conn_config["purity"] = oracledb.PURITY_SELF
elif purity == "default":
conn_config["purity"] = oracledb.PURITY_DEFAULT
expire_time = conn.extra_dejson.get("expire_time")
if expire_time:
conn_config["expire_time"] = expire_time
conn = oracledb.connect(**conn_config) # type: ignore[assignment]
if mod is not None:
conn.module = mod
# if Connection.schema is defined, set schema after connecting successfully
# cannot be part of conn_config
# https://python-oracledb.readthedocs.io/en/latest/api_manual/connection.html?highlight=schema#Connection.current_schema
# Only set schema when not using conn.schema as Service Name
if schema and service_name:
conn.current_schema = schema
return conn