Singer Protocol Implementation
Relevant source files
The following files were used as context for generating this wiki page:
The Singer Protocol implementation in flext-target-oracle provides a robust, type-safe interface for consuming Singer-formatted data and persisting it into an Oracle Database. It functions as a "Target" in the Singer ecosystem, processing standard message types—SCHEMA, RECORD, STATE, and ACTIVATE_VERSION—while leveraging the FLEXT ecosystem's patterns for error handling and database abstraction.
The system is designed with a clear separation of concerns: a high-level client coordinates message routing and state management, while a specialized loader handles the technical details of Oracle DDL/DML operations. This architecture ensures strict adherence to Singer specifications while optimizing for Oracle-specific performance features like bulk operations and specialized data type mapping.
Sources: src/flext_target_oracle/_utilities/client.py:12-25, src/flext_target_oracle/_utilities/loader.py:27-35
Architecture and Components
The implementation follows a layered architecture where the FlextTargetOracle class serves as the public entry point, delegating low-level database operations to the FlextTargetOracleLoader.
Core Class Structure
The diagram shows the relationship between the target client, the database loader, and the configuration settings.
Sources: src/flext_target_oracle/_utilities/client.py:16-25, src/flext_target_oracle/_utilities/loader.py:175-200, src/flext_target_oracle/settings.py:24-30
Key Message Handlers
The implementation uses a match-case pattern to route Singer messages to specific internal handlers:
| Message Type | Handler Method | Purpose |
|---|---|---|
| SCHEMA | _handle_schema | Validates/creates Oracle tables based on JSON schema definitions. |
| RECORD | _handle_record | Buffers data for batch insertion into the destination table. |
| STATE | _handle_state | Updates the internal state tracking for bookmarks and replication keys. |
| ACTIVATE_VERSION | _handle_activate_version | Logs and acknowledges version activation for full-table syncs. |
Sources: src/flext_target_oracle/_utilities/client.py:145-160, src/flext_target_oracle/_utilities/client.py:205-245
Message Processing Flow
The processing of a Singer stream follows a specific sequence where metadata must precede data.
The sequence diagram illustrates the typical lifecycle of a Singer data stream from initial schema definition to final batch persistence.
Sources: src/flext_target_oracle/_utilities/client.py:145-180, src/flext_target_oracle/_utilities/loader.py:440-475, examples/01_basic_usage.py:140-185
Data Loading Strategies
The implementation supports multiple strategies for persisting data, controlled via FlextTargetOracleSettings.
Batching and Buffering
The loader maintains internal buffers for each stream. When the number of records in a buffer reaches the configured batch_size (default: 1000), a flush operation is triggered automatically.
Sources: src/flext_target_oracle/_utilities/loader.py:480-500, src/flext_target_oracle/settings.py:65-68
Persistence Modes
- Flattened (Default): Each property in the Singer schema becomes a column in the Oracle table.
- JSON Storage: Nested data is persisted as a single
CLOBcolumn named according tojson_column_name. - Hybrid: Persists primary keys as columns and the entire record payload in a JSON column.
Sources: src/flext_target_oracle/_utilities/loader.py:93-115, src/flext_target_oracle/settings.py:84-88
SDC (Singer Data Capture) Audit Columns
The loader automatically injects management columns into every table for auditing purposes:
_SDC_EXTRACTED_AT: The timestamp when the data was extracted from the source._SDC_LOADED_AT: The timestamp when the record was persisted in Oracle.
Sources: src/flext_target_oracle/_utilities/loader.py:160-173, src/flext_target_oracle/_utilities/loader.py:60-70
Configuration and Rules
Configuration is managed via a Pydantic-based settings class that enforces business rules and manages Oracle connection parameters.
| Parameter | Type | Default | Description |
|---|---|---|---|
oracle_host | str | localhost | Database host address. |
batch_size | int | 1000 | Records per batch before flushing. |
default_target_schema | str | SINGER_DATA | Target Oracle schema for tables. |
load_method | str | INSERT | Strategy: INSERT, BULK_INSERT, or MERGE. |
use_bulk_operations | bool | True | Whether to use execute_many for performance. |
Sources: src/flext_target_oracle/settings.py:33-110
Business Rule Validation
The settings class includes a validate_business_rules method that ensures:
- Required connection fields (
oracle_host,oracle_service_name) are present. - The
commit_intervaldoes not exceed thebatch_size. - The
default_target_schemais specified.
Sources: src/flext_target_oracle/settings.py:127-140
Protocol and Result Handling
The implementation utilizes Railway-Oriented Programming through the p.Result (or r) type from the FLEXT core. Every major operation returns either an .ok() result containing the value or a .fail() result containing error details.
Implementation Protocols
The module defines formal Python Protocols for various Singer components:
- ConnectionService: For validating connectivity.
- SchemaService: For ensuring table existence.
- BatchService: For queuing and flushing records.
- RecordService: For transforming Singer payloads to Oracle types.
Sources: src/flext_target_oracle/_protocols/base.py:100-145
Finalization Statistics
Upon completion via finalize(), the system returns a LoaderFinalizeResult model containing:
total_records: Sum of all records processed across all streams.streams_processed: Count of distinct streams handled.loading_operation: Detailed status including records successfully loaded vs failed.
Sources: src/flext_target_oracle/_utilities/loader.py:537-558
The Singer Protocol Implementation ensures that high-volume data streams can be reliably mapped from generic JSON formats into structured Oracle schemas while maintaining data integrity through state management and transaction-aware batching.