From 2d23619ee015baa995e1862e2c6e75deb549770b Mon Sep 17 00:00:00 2001 From: Roel de Vries Date: Tue, 17 Oct 2023 22:43:33 +0200 Subject: [PATCH] Use oracledb default setting instead of output type handler --- gobcore/datastore/oracle.py | 22 ++++++++-------------- tests/gobcore/datastore/test_oracle.py | 12 ++---------- 2 files changed, 10 insertions(+), 24 deletions(-) diff --git a/gobcore/datastore/oracle.py b/gobcore/datastore/oracle.py index 6bafcc82..2551c146 100644 --- a/gobcore/datastore/oracle.py +++ b/gobcore/datastore/oracle.py @@ -1,6 +1,6 @@ from pathlib import Path from collections.abc import Iterator, Callable -from typing import Optional, Any +from typing import Any import os import oracledb @@ -28,6 +28,10 @@ """ os.environ["NLS_LANG"] = ".UTF8" +# returns strings or bytes instead of a locator +# https://python-oracledb.readthedocs.io/en/latest/user_guide/lob_data.html#fetching-lobs-as-strings-and-bytes +oracledb.defaults.fetch_lobs = False + class OracleDatastore(SqlDatastore): @@ -73,15 +77,12 @@ def connect(self) -> oracledb.Connection: dsn = self._build_connection_string(**self.connection_config, params=dict(params)) try: - connection = oracledb.connect(dsn) + self.connection = oracledb.connect(dsn) except oracledb.OperationalError as e: raise GOBException( f"Database connection for source {self.connection_config['name']} {self.user} failed: {e}" ) - - connection.outputtypehandler = self._output_type_handler - self.connection = connection - return connection + return self.connection def query(self, query: str, **kwargs) -> Iterator[dict[str, Any]]: """Return query result iterator from the database formatted as dictionary.""" @@ -100,7 +101,7 @@ def execute(self, query: str) -> None: """Executes a SQL statement on the database and commits the changes.""" with self.connection.cursor() as cur: cur.execute(query) - self.connection.commit() + self.connection.commit() def list_tables_for_schema(self, schema: str) -> list[str]: raise NotImplementedError("Please implement list_tables_for_schema for OracleDatastore") @@ -125,10 +126,3 @@ def _dict_cursor(cursor: oracledb.Cursor) -> Callable[..., dict[str, Any]]: """Convert query tuple a dictionary with column names as keys.""" cols = [d[0].lower() for d in cursor.description] return lambda *args: dict(zip(cols, args)) - - @staticmethod - def _output_type_handler( - cursor: oracledb.Cursor, name: str, default_type, size, precision, scale - ) -> Optional[oracledb.Var]: - if default_type == oracledb.CLOB: - return cursor.var(oracledb.LONG_STRING, arraysize=cursor.arraysize) diff --git a/tests/gobcore/datastore/test_oracle.py b/tests/gobcore/datastore/test_oracle.py index f5d27fbe..d74d8376 100644 --- a/tests/gobcore/datastore/test_oracle.py +++ b/tests/gobcore/datastore/test_oracle.py @@ -38,6 +38,8 @@ def test_init(self, mock_temp, mock_path, mock_init_client): OracleDatastore(self.config) mock_init_client.assert_not_called() + assert oracledb.defaults.fetch_lobs is False + def test_init_missing_keys(self): with self.assertRaisesRegex(GOBException, "database,password,port,username"): OracleDatastore({"name": "my config", "host": "my_host"}) @@ -50,7 +52,6 @@ def test_connect(self, mock_connect): "username/password@tcp://host1:1234,host2:1234/SID?" "retry_count=3&connection_timeout=3&failover=on&load_balance=off" ) - assert self.store.connection.outputtypehandler == self.store._output_type_handler assert self.store.connection == conn == mock_connect.return_value # single host @@ -69,15 +70,6 @@ def test_connect_opserror(self, _): def test_makedict(self, mock_cursor): assert {"a": 1, "b": 2, "c": 3} == self.store._dict_cursor(mock_cursor)(*(1, 2, 3, 4)) - @patch("oracledb.Cursor", autospec=True, spec_set=True, arraysize=5) - def test_output_type_handler(self, mock_cursor): - self.store._output_type_handler(mock_cursor, "name", oracledb.CLOB, 0, 0, 0) - mock_cursor.var.assert_called_with(oracledb.LONG_STRING, arraysize=5) - - # some other type - result = self.store._output_type_handler(mock_cursor, "name", oracledb.NUMBER, 0, 0, 0) - assert result is None - @patch("oracledb.Cursor", autospec=True, spec_set=True, arraysize=5, description=(("id",),)) @patch("oracledb.Connection", autospec=True, spec_set=True) def test_query(self, mock_conn, mock_cursor):