/flext-sh/flext-tap-oracle

Wiki directory

Browse AI-generated documentation for any repository.

Stream Definitions

Relevant source files

The following files were used as context for generating this wiki page:

Stream Definitions in the flext-tap-oracle project define how Oracle database objects (tables and views) are mapped to Singer-compatible streams. This system leverages the flext-db-oracle API to perform schema introspection, data extraction, and type transformation, ensuring that Oracle-specific data types are correctly represented in JSON-based Singer messages. The architecture is designed for high performance, utilizing connection pooling and optimized fetch sizes for large-scale enterprise data movement.

The primary objective of these definitions is to provide a unified interface for data extraction that handles metadata management, row count estimation, and record transformation. By integrating with flext-meltano and flext-core, the stream definitions support both FULL_TABLE and INCREMENTAL replication methods, governed by a set of comprehensive configuration rules defined in the project's orchestration layer.

Sources: [src/flext_tap_oracle/streams.py:12-25](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L12-L25), [src/flext_tap_oracle/rules.json:134-142](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L134-L142)

Stream Architecture and Implementation

The project implements a core OracleStream class which serves as the execution engine for data extraction. It relies on the FlextDbOracleApi for all database interactions, avoiding direct dependencies on lower-level libraries like SQLAlchemy within the stream logic itself.

OracleStream Components

The OracleStream handles the lifecycle of a stream, from initialization with a validated schema to the execution of SQL queries and subsequent record transformation.

Row Count Estimation: Uses Oracle system views and SELECT COUNT(*) queries to estimate the volume of data for progress tracking.
Metadata Management: Retrieves table metadata including column definitions, schema names, and table types.
Record Processing: Iterates through query results, transforming complex Oracle types (like CLOBs, BLOBs, and Dates) into Singer-compliant JSON formats.

Sources: [src/flext_tap_oracle/streams.py:27-60](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L27-L60), [src/flext_tap_oracle/streams.py:99-130](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L99-L130)

Stream Data Flow

The following diagram illustrates the flow of data from the Oracle database through the Stream implementation to the Singer output.

Generating architecture diagram...

The diagram shows the extraction sequence where metadata is used to guide the transformation of raw Oracle rows into formatted Singer records.
Sources: [src/flext_tap_oracle/streams.py:69-97](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L69-L97), [src/flext_tap_oracle/streams.py:159-204](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L159-L204)

Discovery and Factory Methods

Stream instances are created dynamically during the discovery phase. The StreamFactory provides methods to instantiate streams based on metadata retrieved from the database.

Stream Creation Logic

Table Discovery: The FlextTapOracleUtilitiesClientMixin uses the Oracle API to fetch available tables in a target schema (defaulting to USER).
Factory Instantiation: The StreamFactory maps discovered tables to OracleStream instances, applying a configured stream_prefix.
Schema Generation: Standard Singer properties are generated, defining the object structure and data types for the catalog.

Sources: [src/flext_tap_oracle/_utilities/client.py:28-56](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/_utilities/client.py#L28-L56), [src/flext_tap_oracle/streams.py:206-254](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L206-L254)

Type Mapping

Oracle data types are mapped to Singer types according to the project rules.

Oracle Type	Singer Type	Transformation Note
VARCHAR2 / CHAR	string	Standard string
NUMBER	number	Preserves precision
DATE / TIMESTAMP	string	Converted to ISO-8601 format
CLOB / BLOB	string	Stringified representation
JSON	t.JsonValue	Native JSON support

Sources: [src/flext_tap_oracle/rules.json:164-188](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L164-L188), [src/flext_tap_oracle/streams.py:180-202](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L180-L202)

Configuration and Rules

Streams are governed by configuration_rules defined in rules.json and validated via Pydantic models in settings.py.

Generating architecture diagram...

This class diagram highlights the relationship between runtime settings and the discovery configuration that defines stream boundaries.
Sources: [src/flext_tap_oracle/settings.py:25-58](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/settings.py#L25-L58), [src/flext_tap_oracle/rules.json:144-150](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L144-L150)

Replication Methods

The system supports two primary replication strategies:

FULL_TABLE: The entire table is extracted in every sync. No state bookmarking is required.
INCREMENTAL: Uses a replication_key (typically a timestamp or number) to extract only new or updated records since the last sync.

Sources: [src/flext_tap_oracle/rules.json:115-120](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L115-L120), [src/flext_tap_oracle/rules.json:152-160](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L152-L160)

Performance and Extraction Metrics

The OracleStream tracks execution statistics through OracleTapExecutionStats. This includes monitoring record counts, byte volume, and processing efficiency.

Metric Definitions

Metric Field	Description	Source
`total_records`	Count of all records successfully extracted.	`models.py`
`oracle_query_time`	Total time spent waiting for Oracle query responses.	`models.py`
`avg_records_per_second`	Throughput measurement for the extraction process.	`models.py`
`batch_size`	Configurable chunk size for fetching records (default 1000).	`rules.json`

Sources: [src/flext_tap_oracle/models.py:86-105](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/models.py#L86-L105), [src/flext_tap_oracle/rules.json:69-72](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L69-L72)

Performance Optimization

To ensure efficient extraction from Oracle, the streams utilize:

Cursor Array Size: Defaulted to 100 for memory management.
Fetch Size: Defaulted to 1000 to balance network latency and memory usage.
Parallel Streams: Support for multiple parallel extractions (recommended max of 4).

Sources: [src/flext_tap_oracle/rules.json:200-210](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L200-L210), [src/flext_tap_oracle/rules.json:237-240](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/rules.json#L237-L240)

Record Transformation Logic

The _transform_oracle_types method is critical for Singer compliance. It ensures that Oracle's internal representations of time and large objects are serialized correctly for downstream consumers.

# From src/flext_tap_oracle/streams.py
def _transform_oracle_types(
    self,
    record: t.JsonMapping,
    column_metadata: t.SequenceOf[m.DbOracle.ColumnMetadata],
) -> t.JsonMapping:
    # Logic to map Oracle types to JSON-safe formats
    for column_name, value in record.items():
        if value is None:
            transformed_record[column_name] = None
            continue
        # ... type detection via metadata ...
        if oracle_type_str.upper().startswith(("DATE", "TIMESTAMP")):
            if isinstance(value, datetime):
                transformed_record[column_name] = value.isoformat()
        elif oracle_type_str.upper().startswith(("CLOB", "BLOB")):
            transformed_record[column_name] = str(value)

Sources: [src/flext_tap_oracle/streams.py:180-202](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L180-L202)

Stream Definitions

Relevant source files

The following files were used as context for generating this wiki page:

Stream Architecture and Implementation

OracleStream Components

The OracleStream handles the lifecycle of a stream, from initialization with a validated schema to the execution of SQL queries and subsequent record transformation.

Row Count Estimation: Uses Oracle system views and SELECT COUNT(*) queries to estimate the volume of data for progress tracking.
Metadata Management: Retrieves table metadata including column definitions, schema names, and table types.
Record Processing: Iterates through query results, transforming complex Oracle types (like CLOBs, BLOBs, and Dates) into Singer-compliant JSON formats.

Stream Data Flow

The following diagram illustrates the flow of data from the Oracle database through the Stream implementation to the Singer output.

Generating architecture diagram...

Discovery and Factory Methods

Stream instances are created dynamically during the discovery phase. The StreamFactory provides methods to instantiate streams based on metadata retrieved from the database.

Stream Creation Logic

Table Discovery: The FlextTapOracleUtilitiesClientMixin uses the Oracle API to fetch available tables in a target schema (defaulting to USER).
Factory Instantiation: The StreamFactory maps discovered tables to OracleStream instances, applying a configured stream_prefix.
Schema Generation: Standard Singer properties are generated, defining the object structure and data types for the catalog.

Type Mapping

Oracle data types are mapped to Singer types according to the project rules.

Oracle Type	Singer Type	Transformation Note
VARCHAR2 / CHAR	string	Standard string
NUMBER	number	Preserves precision
DATE / TIMESTAMP	string	Converted to ISO-8601 format
CLOB / BLOB	string	Stringified representation
JSON	t.JsonValue	Native JSON support

Configuration and Rules

Streams are governed by configuration_rules defined in rules.json and validated via Pydantic models in settings.py.

Generating architecture diagram...

Replication Methods

The system supports two primary replication strategies:

FULL_TABLE: The entire table is extracted in every sync. No state bookmarking is required.
INCREMENTAL: Uses a replication_key (typically a timestamp or number) to extract only new or updated records since the last sync.

Performance and Extraction Metrics

The OracleStream tracks execution statistics through OracleTapExecutionStats. This includes monitoring record counts, byte volume, and processing efficiency.

Metric Definitions

Metric Field	Description	Source
`total_records`	Count of all records successfully extracted.	`models.py`
`oracle_query_time`	Total time spent waiting for Oracle query responses.	`models.py`
`avg_records_per_second`	Throughput measurement for the extraction process.	`models.py`
`batch_size`	Configurable chunk size for fetching records (default 1000).	`rules.json`

Performance Optimization

To ensure efficient extraction from Oracle, the streams utilize:

Cursor Array Size: Defaulted to 100 for memory management.
Fetch Size: Defaulted to 1000 to balance network latency and memory usage.
Parallel Streams: Support for multiple parallel extractions (recommended max of 4).

Record Transformation Logic

The _transform_oracle_types method is critical for Singer compliance. It ensures that Oracle's internal representations of time and large objects are serialized correctly for downstream consumers.

# From src/flext_tap_oracle/streams.py
def _transform_oracle_types(
    self,
    record: t.JsonMapping,
    column_metadata: t.SequenceOf[m.DbOracle.ColumnMetadata],
) -> t.JsonMapping:
    # Logic to map Oracle types to JSON-safe formats
    for column_name, value in record.items():
        if value is None:
            transformed_record[column_name] = None
            continue
        # ... type detection via metadata ...
        if oracle_type_str.upper().startswith(("DATE", "TIMESTAMP")):
            if isinstance(value, datetime):
                transformed_record[column_name] = value.isoformat()
        elif oracle_type_str.upper().startswith(("CLOB", "BLOB")):
            transformed_record[column_name] = str(value)

Sources: [src/flext_tap_oracle/streams.py:180-202](https://github.com/flext-sh/flext-tap-oracle/blob/fda799d0602e48c3fc630912b6c14feb9a1d4a85/src/flext_tap_oracle/streams.py#L180-L202)