singer_sdk.connectors.sql.JSONSchemaToSQL

class singer_sdk.connectors.sql.JSONSchemaToSQL[source]

A configurable mapper for converting JSON Schema types to SQLAlchemy types.

This class provides a mapping from JSON Schema types to SQLAlchemy types.

Added in version 0.42.0.

Changed in version 0.44.0: Added the singer_sdk.connectors.sql.JSONSchemaToSQL.register_sql_datatype_handler() method to map custom x-sql-datatype annotations into SQLAlchemy types.

classmethod from_config(config, *, max_varchar_length)[source]

Create a new instance from a configuration dictionary.

Override this to instantiate this converter with values from the target’s configuration dictionary.

class CustomJSONSchemaToSQL(JSONSchemaToSQL):
    @classmethod
    def from_config(cls, config, **kwargs):
        return cls(max_varchar_length=config.get("max_varchar_length"))
Parameters:

  • config (Mapping[str, t.Any]) – The configuration dictionary.

  • max_varchar_length (int | None) – The absolute maximum length for VARCHAR columns that the database supports.

Returns:

A new instance of the class.

Return type:

JSONSchemaToSQL

__init__(*, max_varchar_length=None)[source]

Initialize the mapper with default type mappings.

Parameters:

max_varchar_length (int | None) – The absolute maximum length for VARCHAR columns that the database supports.

Return type:

None

handle_multiple_types(types)[source]

Handle multiple types by returning a VARCHAR.

Parameters:

types (Sequence[str]) – The list of types to handle.

Returns:

A VARCHAR type.

Return type:

TypeEngine

handle_raw_string(schema)[source]

Handle a string type generically.

Parameters:

schema (dict) – The JSON Schema object.

Returns:

Appropriate SQLAlchemy type.

Return type:

TypeEngine

register_format_handler(format_name, handler)[source]

Register a custom format handler.

Parameters:

  • format_name (str) – The format string (e.g., “date-time”, “email”, “custom-format”).

  • handler (type[TypeEngine] | Callable[[dict], TypeEngine]) – Either a SQLAlchemy type class or a callable that takes a schema dict and returns a SQLAlchemy type instance.

Return type:

None

register_sql_datatype_handler(sql_datatype, handler)[source]

Register a custom x-sql-datatype handler.

Parameters:

  • sql_datatype (str) – The x-sql-datatype string.

  • handler (type[TypeEngine] | Callable[[dict], TypeEngine]) – Either a SQLAlchemy type class or a callable that takes a schema dict and returns a SQLAlchemy type instance.

Return type:

None

Example

>>> from sqlalchemy.types import SMALLINT
>>> to_sql = JSONSchemaToSQL()
>>> to_sql.register_sql_datatype_handler("smallint", SMALLINT)
register_type_handler(json_type, handler)[source]

Register a custom type handler.

Parameters:

  • json_type (str) – The JSON Schema type to handle.

  • handler (type[TypeEngine] | Callable[[dict], TypeEngine]) – Either a SQLAlchemy type class or a callable that takes a schema dict and returns a SQLAlchemy type instance.

Return type:

None

to_sql_type(schema)[source]

Convert a JSON Schema type definition to a SQLAlchemy type.

Parameters:

schema (dict) – The JSON Schema object.

Returns:

The corresponding SQLAlchemy type.

Return type:

TypeEngine

property fallback_type: type[TypeEngine][source]

Return the fallback type.

Returns:

The fallback type.