Singer Integration
Relevant source files
The following files were used as context for generating this wiki page:
Singer Integration within the FLEXT DB Oracle library provides a specialized service layer for mapping Singer JSON schemas to Oracle-compatible data types. This functionality is essential for data engineering workflows that utilize the Singer tap/target standard, allowing for seamless transformation of JSON-based stream definitions into structured Oracle database tables.
The integration is primarily implemented as a mixin service, FlextDbOracleServiceSinger, which handles the conversion logic for individual types and the mapping of entire schema properties. This service ensures that standard Singer types like integer, number, and boolean are correctly represented as Oracle NUMBER variants, while handling specialized formats like date-time.
Sources: src/flext_db_oracle/services/singer.py:1-15, src/flext_db_oracle/models.py:345-365
Architecture and Components
The Singer integration follows the project's Clean Architecture pattern, residing within the Service layer as a mixin that is composed into the larger FlextDbOracleServices facade.
Core Components
| Component | Responsibility |
|---|---|
FlextDbOracleServiceSinger | Service mixin containing the logic for type conversion and schema mapping. |
SingerSchema | A domain model representing the incoming Singer property definitions. |
SingerField | A domain model representing an individual field's type definition within a schema. |
TypeMapping | A domain model representing the resulting Oracle type map after conversion. |
Sources: src/flext_db_oracle/services/singer.py:24-28, src/flext_db_oracle/models.py:345-365
Data Flow Diagram
The following diagram illustrates the flow of data from a raw Singer JSON schema through the normalization and conversion logic to produce an Oracle type mapping.
The workflow begins by validating whether the input is a pre-validated model or a raw JSON mapping. It then normalizes the fields and iterates through them to perform individual type conversions.
Sources: src/flext_db_oracle/services/singer.py:48-91
Type Mapping Logic
The mapping logic translates JSON Schema types used by Singer into Oracle SQL data types. This is handled by the convert_singer_type method, which supports both single strings and sequences of strings (standard in Singer for handling nullable types).
Type Conversion Table
| Singer Type | Oracle Type | Notes |
|---|---|---|
string | VARCHAR2(4000) | Default mapping for strings. |
integer | NUMBER(38) | Maps to Oracle's high-precision integer representation. |
number | NUMBER | Generic numeric type. |
boolean | NUMBER(1) | Oracle typically uses 0/1 for booleans. |
date-time | TIMESTAMP | Triggered by the format hint in the schema. |
| Default | VARCHAR2(255) | Fallback for unrecognized types. |
Sources: src/flext_db_oracle/services/singer.py:30-46, tests/unit/test_services.py:112-126
Schema Mapping Implementation
The map_singer_schema function processes a complex Singer schema object. It extracts the properties dictionary and converts each field. If a field definition contains a format key with the value date-time, the system overrides the base type to use TIMESTAMP.
def convert_singer_type(
self,
singer_type: str | t.StrSequence = "string",
_format_hint: str | None = None,
) -> p.Result[str]:
singer_type = self._normalize_singer_type(singer_type)
if _format_hint == "date-time":
return r[str].ok("TIMESTAMP")
type_map = {
"string": "VARCHAR2(4000)",
"integer": "NUMBER(38)",
"number": "NUMBER",
"boolean": "NUMBER(1)",
"date-time": "TIMESTAMP",
}
oracle_type = type_map.get(singer_type, "VARCHAR2(255)")
return r[str].ok(oracle_type)
Sources: src/flext_db_oracle/services/singer.py:30-46
Integration with Public API
The FlextDbOracleApiRuntime exposes Singer integration methods to the end-user. This allows developers to interact with the Singer service via the main FlextDbOracleApi facade.
Sequence Diagram: API Type Conversion
The following sequence diagram shows how a user request for type conversion is dispatched through the API runtime to the underlying Singer service.
The API runtime acts as a pass-through and result formatter, converting internal domain models back into standard Python dictionaries for the user.
Sources: src/flext_db_oracle/services/api_runtime.py:180-190, src/flext_db_oracle/services/api_runtime.py:240-250
Implementation Details in Models
The Singer integration relies on specific domain models defined in FlextDbOracleModels.DbOracle. These models enforce validation using Pydantic, ensuring that schema properties are correctly structured before conversion.
SingerField: Validates thetypeattribute, which can be a single string or a sequence (e.g.,["string", "null"]).SingerSchema: Contains a mapping of field names toSingerFieldinstances.TypeMapping: Encapsulates the final dictionary of Oracle column names and their corresponding Oracle types.
Sources: src/flext_db_oracle/models.py:345-365
Testing and Validation
The integration is verified through both unit and end-to-end tests to ensure type conversion accuracy.
Unit Test Verification
Unit tests in test_services.py validate that specific Singer inputs result in the correct Oracle SQL strings. For instance, an input of ["string", "null"] correctly normalizes to VARCHAR2(4000).
Sources: tests/unit/test_services.py:112-146
E2E Validation
End-to-end tests in test_oracle.py simulate a real Oracle environment, confirming that the generated schema mappings can be used to describe valid Oracle table structures.
Sources: tests/e2e/test_oracle.py:133-167
Summary
Singer Integration provides the necessary translation layer between the Singer JSON Schema standard and Oracle Database types. By centralizing this logic in FlextDbOracleServiceSinger, the library offers a robust and predictable way to define Oracle tables based on dynamic stream metadata, supporting standard types, nullable fields, and specialized timestamp formats.