feat(elt-pipelines): ISISUSERDB oracle sources pipeline

elt-common/src/elt_common/sources/sqldatabase/__init__.py

@@ -4,7 +4,7 @@
 import logging
 from abc import abstractmethod
 from collections.abc import Callable, Generator, Iterable, Iterator
-from typing import NamedTuple
+from typing import Any, NamedTuple
 
 import pyarrow as pa
 import pyarrow.compute as pc
@@ -191,6 +191,21 @@ def extractor(watermark, *, _name=name):
 
             yield resource_name, properties
 
+    def _columns_for_select(self, table: sa.Table) -> Iterable[Any]:
+        """Retrieve columns for selection applying any DB-specific requirements."""
+        if self._engine.dialect.name == "oracle":
+            # Apply UTC extraction strictly to Oracle databases to avoid Thin mode DPY-3022 errors
+            selected_cols = []
+            for col in table.columns:
+                col_type_str = str(col.type).upper()
+                if getattr(col.type, "timezone", False) or "WITH TIME ZONE" in col_type_str:
+                    selected_cols.append(sa.func.sys_extract_utc(col).label(col.name))
+                else:
+                    selected_cols.append(col)
+            return selected_cols
+        else:
+            return table.columns
+
     def _extract_table(
         self,
         name: str,
@@ -205,7 +220,8 @@ def _extract_table(
             self._metadata,
             autoload_with=self._engine,
         )
-        query = sa.select(table)
+
+        query = sa.select(*self._columns_for_select(table))
         if watermark is not None:
             column, max_value = watermark.column, watermark.value
             LOGGER.debug(f"Cursor value detected. Limiting query to {column} > {max_value}")
@@ -214,11 +230,9 @@ def _extract_table(
         if query_filter:
             query = query_filter(query)
 
-        query = query.limit(self.config.row_limit)
+        if self.config.row_limit:
+            query = query.limit(self.config.row_limit)
 
-        # If all the values in a column are null pyarrow won't know what type
-        # the column should be, so we need to explicitly create a schema from
-        # the table
         pa_schema = to_pyarrow_schema(table)
         result = conn.execution_options(yield_per=self.config.chunk_size).execute(query)
 

elt-common/src/elt_common/sources/sqldatabase/schema.py

@@ -6,6 +6,7 @@
 import pyarrow as pa
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from sqlalchemy.dialects.oracle import RAW
 
 
 def to_pyarrow_schema(table: sa.Table) -> pa.Schema:
@@ -63,6 +64,7 @@ def _to_pyarrow_field(column: sa.Column) -> pa.Field:
     sa.VARCHAR: _SQL_ROOT_TYPES[sa.String],
     postgresql.JSON: _SQL_ROOT_TYPES[sa.JSON],
     postgresql.JSONB: _SQL_ROOT_TYPES[sa.JSON],
+    RAW: _SQL_ROOT_TYPES[sa.LargeBinary],
 }
 
 _SQL_TYPE_MAP = _SQL_ROOT_TYPES | _EXTENDED_SQL_TYPES

elt-pipelines/fase/ingest/fase/isisuserdb/isisuserdb.py

@@ -0,0 +1,36 @@
+import sqlalchemy as sa
+from elt_common.extract import ResourceWriteProperties
+from elt_common.sources.sqldatabase import (
+    SqlDatabaseExtract,
+    SqlDatabaseSourceConfig,
+    TableInfo,
+)
+
+
+class PipelineOracleConfig(SqlDatabaseSourceConfig):
+    drivername: str = "oracle+oracledb"
+    tables: list[str]
+
+    @property
+    def connection_url(self) -> sa.URL:
+        return sa.URL.create(
+            drivername=self.drivername,
+            username=self.username,
+            password=self.password.get_secret_value() if self.password else None,
+            host=self.host,
+            port=self.port,
+            query={"service_name": self.database},
+        )
+
+
+class Extract(SqlDatabaseExtract):
+    config_cls = PipelineOracleConfig
+
+    def table_info(self) -> dict[str, TableInfo | None]:
+        """Defines the target tables and their ingestion strategy."""
+        return {
+            table_name: TableInfo(
+                write_properties=ResourceWriteProperties(write_mode="replace")
+            )
+            for table_name in self.config.tables
+        }

elt-pipelines/pyproject.toml

@@ -13,6 +13,11 @@ dependencies = [
 proposal = [
     "sqlalchemy[postgresql-psycopgbinary]>=2.0.0",
 ]
+
+isisuserdb = [
+    "sqlalchemy[oracle-oracledb]>=2.0.0",
+]
+
 statusdisplay = [
     "requests>=2.34.2",
 ]