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_schema(schema_name: str) None
Create target schema.
- Parameters
schema_name – The target schema to create.
- 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:
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_column_add_ddl(table_name: str, column_name: str, column_type: TypeEngine) DDL
Get the create column DDL statement.
Override this if your database uses a different syntax for creating columns.
- Parameters
table_name – Fully qualified table name of column to alter.
column_name – Column name to create.
column_type – New column sqlalchemy type.
- Returns
A sqlalchemy DDL instance.
- static get_column_alter_ddl(table_name: str, column_name: str, column_type: TypeEngine) DDL
Get the alter column DDL statement.
Override this if your database uses a different syntax for altering columns.
- Parameters
table_name – Fully qualified table name of column to alter.
column_name – Column name to alter.
column_type – New column type string.
- Returns
A sqlalchemy DDL instance.
- static get_column_rename_ddl(table_name: str, column_name: str, new_column_name: str) DDL
Get the create column DDL statement.
Override this if your database uses a different syntax for renaming columns.
- Parameters
table_name – Fully qualified table name of column to alter.
column_name – Existing column name.
new_column_name – New column name.
- Returns
A sqlalchemy DDL instance.
- static get_fully_qualified_name(table_name: str | None = None, 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 all 3 name parts not supplied.
- 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, column_names: list[str] | None = None) sqlalchemy.Table
Return a table object.
- Parameters
full_table_name – Fully qualified table name.
column_names – A list of column names to filter to.
- Returns
A table object with column list.
- get_table_columns(full_table_name: str, column_names: list[str] | None = None) dict[str, sqlalchemy.Column]
Return a list of table columns.
- Parameters
full_table_name – Fully qualified table name.
column_names – A list of column names to filter to.
- 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_schema(schema_name: str) None
Create the target database schema.
- Parameters
schema_name – The target schema name.
- 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.
- schema_exists(schema_name: str) bool
Determine if the target database schema already exists.
- Parameters
schema_name – The target database schema name.
- Returns
True if the database schema exists, False if not.
- 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.