singer_sdk.SQLConnector

class singer_sdk.SQLConnector

Base class for SQLAlchemy-based connectors.

The connector class serves as a wrapper around the SQL connection.

The functions of the connector are:

  • connecting to the source

  • generating SQLAlchemy connection and engine objects

  • discovering schema catalog entries

  • performing type conversions to/from JSONSchema types

  • dialect-specific functions, such as escaping and fully qualified names

__init__(config: dict | None = None, sqlalchemy_url: str | None = None) None

Initialize the SQL connector.

Parameters
  • config – The parent tap or target object’s config.

  • sqlalchemy_url – Optional URL for the connection.

column_exists(full_table_name: str, column_name: str) bool

Determine if the target table already exists.

Parameters
  • full_table_name – the target table name.

  • column_name – the target column name.

Returns

True if table exists, False if not.

property config: dict

If set, provides access to the tap or target config.

Returns

The settings as a dict.

property connection: Connection

Return or set the SQLAlchemy connection object.

Returns

The active SQLAlchemy connection object.

create_empty_table(full_table_name: str, schema: dict, primary_keys: list[str] | None = None, partition_keys: list[str] | None = None, as_temp_table: bool = False) None

Create an empty target table.

Parameters
  • full_table_name – the target table name.

  • schema – the JSON schema for the new table.

  • primary_keys – list of key properties.

  • partition_keys – list of partition keys.

  • as_temp_table – True to create a temp table.

Raises
  • NotImplementedError – if temp tables are unsupported and as_temp_table=True.

  • RuntimeError – if a variant schema is passed with no properties defined.

create_sqlalchemy_connection() Connection

Return a new SQLAlchemy connection using the provided config.

By default this will create using the sqlalchemy stream_results=True option described here: https://docs.sqlalchemy.org/en/14/core/connections.html#using-server-side-cursors-a-k-a-stream-results

Developers may override this method if their provider does not support server side cursors (stream_results) or in order to use different configurations options when creating the connection object.

Returns

A newly created SQLAlchemy engine object.

create_sqlalchemy_engine() Engine

Return a new SQLAlchemy engine using the provided config.

Developers can generally override just one of the following: sqlalchemy_engine, sqlalchemy_url`.

Returns

A newly created SQLAlchemy engine object.

discover_catalog_entries() list[dict]

Return a list of catalog entries from discovery.

Returns

The discovered catalog entries as a list.

discover_catalog_entry(engine: Engine, inspected: Inspector, schema_name: str, table_name: str, is_view: bool) CatalogEntry

Create CatalogEntry object for the given table or a view.

Parameters
  • engine – SQLAlchemy engine

  • inspected – SQLAlchemy inspector instance for engine

  • schema_name – Schema name to inspect

  • table_name – Name of the table or a view

  • is_view – Flag whether this object is a view, returned by get_object_names

Returns

CatalogEntry object for the given table or a view

static get_fully_qualified_name(table_name: str, schema_name: str | None = None, db_name: str | None = None, delimiter: str = '.') str

Concatenates a fully qualified name from the parts.

Parameters
  • table_name – The name of the table.

  • schema_name – The name of the schema. Defaults to None.

  • db_name – The name of the database. Defaults to None.

  • delimiter – Generally: ‘.’ for SQL names and ‘-’ for Singer names.

Raises

ValueError – If table_name is not provided or if neither schema_name or db_name are provided.

Returns

The fully qualified name as a string.

get_object_names(engine: Engine, inspected: Inspector, schema_name: str) list[tuple[str, bool]]

Return a list of syncable objects.

Parameters
  • engine – SQLAlchemy engine

  • inspected – SQLAlchemy inspector instance for engine

  • schema_name – Schema name to inspect

Returns

List of tuples (<table_or_view_name>, <is_view>)

get_schema_names(engine: Engine, inspected: Inspector) list[str]

Return a list of schema names in DB.

Parameters
  • engine – SQLAlchemy engine

  • inspected – SQLAlchemy inspector instance for engine

Returns

List of schema names

get_sqlalchemy_url(config: dict[str, Any]) str

Return the SQLAlchemy URL string.

Developers can generally override just one of the following: sqlalchemy_engine, get_sqlalchemy_url.

Parameters

config – A dictionary of settings from the tap or target config.

Returns

The URL as a string.

Raises

ConfigValidationError – If no valid sqlalchemy_url can be found.

get_table(full_table_name: str) Table

Return a table object.

Parameters

full_table_name – Fully qualified table name.

Returns

A table object with column list.

get_table_columns(full_table_name: str) dict[str, sqlalchemy.Column]

Return a list of table columns.

Parameters

full_table_name – Fully qualified table name.

Returns

An ordered list of column objects.

property logger: Logger

Get logger.

Returns

Plugin logger.

merge_sql_types(sql_types: list[sqlalchemy.types.TypeEngine]) sqlalchemy.types.TypeEngine

Return a compatible SQL type for the selected type list.

Parameters

sql_types – List of SQL types.

Returns

A SQL type that is compatible with the input types.

Raises

ValueError – If sql_types argument has zero members.

parse_full_table_name(full_table_name: str) tuple[str | None, str | None, str]

Parse a fully qualified table name into its parts.

Developers may override this method if their platform does not support the traditional 3-part convention: db_name.schema_name.table_name

Parameters

full_table_name – A table name or a fully qualified table name. Depending on SQL the platform, this could take the following forms: - <db>.<schema>.<table> (three part names) - <db>.<table> (platforms which do not use schema groupings) - <schema>.<name> (if DB name is already in context) - <table> (if DB name and schema name are already in context)

Returns

A three part tuple (db_name, schema_name, table_name) with any unspecified or unused parts returned as None.

prepare_column(full_table_name: str, column_name: str, sql_type: TypeEngine) None

Adapt target table to provided schema if possible.

Parameters
  • full_table_name – the target table name.

  • column_name – the target column name.

  • sql_type – the SQLAlchemy type.

prepare_table(full_table_name: str, schema: dict, primary_keys: list[str], partition_keys: list[str] | None = None, as_temp_table: bool = False) None

Adapt target table to provided schema if possible.

Parameters
  • full_table_name – the target table name.

  • schema – the JSON Schema for the table.

  • primary_keys – list of key properties.

  • partition_keys – list of partition keys.

  • as_temp_table – True to create a temp table.

quote(name: str) str

Quote a name if it needs quoting, using ‘.’ as a name-part delimiter.

Examples

“my_table” => “my_table” “my_schema.my_table” => “my_schema.`my_table`”

Parameters

name – The unquoted name.

Returns

The quoted name.

Return type

str

rename_column(full_table_name: str, old_name: str, new_name: str) None

Rename the provided columns.

Parameters
  • full_table_name – The fully qualified table name.

  • old_name – The old column to be renamed.

  • new_name – The new name for the column.

Raises

NotImplementedError – If self.allow_column_rename is false.

property sqlalchemy_url: str

Return the SQLAlchemy URL string.

Returns

The URL as a string.

table_exists(full_table_name: str) bool

Determine if the target table already exists.

Parameters

full_table_name – the target table name.

Returns

True if table exists, False if not, None if unsure or undetectable.

static to_jsonschema_type(sql_type: str | sqlalchemy.types.TypeEngine | type[sqlalchemy.types.TypeEngine] | Any) dict

Return a JSON Schema representation of the provided type.

By default will call typing.to_jsonschema_type() for strings and SQLAlchemy types.

Developers may override this method to accept additional input argument types, to support non-standard types, or to provide custom typing logic.

Parameters

sql_type – The string representation of the SQL type, a SQLAlchemy TypeEngine class or object, or a custom-specified object.

Raises

ValueError – If the type received could not be translated to jsonschema.

Returns

The JSON Schema representation of the provided type.

static to_sql_type(jsonschema_type: dict) TypeEngine

Return a JSON Schema representation of the provided type.

By default will call typing.to_sql_type().

Developers may override this method to accept additional input argument types, to support non-standard types, or to provide custom typing logic.

If overriding this method, developers should call the default implementation from the base class for all unhandled cases.

Parameters

jsonschema_type – The JSON Schema representation of the source type.

Returns

The SQLAlchemy type representation of the data type.