bi_etl.database.database_metadata module

Created on Dec 23, 2015

@author: Derek Wood

class bi_etl.database.database_metadata.DatabaseMetadata(bind=None, reflect=False, schema=None, quote_schema=None, naming_convention={'ix': 'ix_%(column_0_label)s'}, info=None, database_name=None, uses_bytes_length_limits=None)[source]

Bases: MetaData

A light wrapper over sqlalchemy.schema.MetaData

__init__(bind=None, reflect=False, schema=None, quote_schema=None, naming_convention={'ix': 'ix_%(column_0_label)s'}, info=None, database_name=None, uses_bytes_length_limits=None)[source]

Initialize database metadata wrapper.

Parameters:

  • bind (sqlalchemy.engine.Engine, optional) – Database engine to bind to

  • reflect (bool, optional) – Whether to reflect database schema, by default False

  • schema (str, optional) – Default schema name

  • quote_schema (bool, optional) – Whether to quote schema names

  • naming_convention (dict, optional) – Naming convention for constraints

  • info (dict, optional) – Optional metadata dictionary

  • database_name (str, optional) – Name of the database

  • uses_bytes_length_limits (bool, optional) – Whether database uses byte-based column length limits

begin(connection_name: str | None = None) → Transaction[source]

Begin a new transaction or return the active transaction.

Parameters:

connection_name (str, optional) – Name of the connection

Returns:

Active transaction

Return type:

sqlalchemy.engine.base.Transaction

clear() → None

Clear all Table objects from this MetaData.

close_connection(connection_name: str | None = None)[source]

Close a specific connection.

Parameters:

connection_name (str, optional) – Name of the connection to close

close_connections(exceptions: set | None = None)[source]

Close all connections except those in the exceptions set.

Parameters:

exceptions (set, optional) – Set of connection names to exclude from closing

commit(connection_name: str | None = None)[source]

Commit based on a connection name rather than via a ‘sqlalchemy.engine.base.Transaction’ object (which you could call .commit() on)

Parameters:

connection_name

connect(connection_name: str | None = None) → Connection[source]

Open or retrieve (if already open) a database connection.

Parameters:

connection_name (str, optional) – Name of the connection

Returns:

Database connection

Return type:

sqlalchemy.engine.base.Connection

connection(connection_name: str | None = None, open_if_not_exist: bool = True, open_if_closed: bool = True) → Connection[source]

Get or create a database connection from the pool by name (with a default).

Parameters:

  • connection_name (str, optional) – Name of the connection to retrieve. Default is used if not provided.

  • open_if_not_exist (bool, optional) – Create connection if it doesn’t exist, by default True

  • open_if_closed (bool, optional) – Reopen connection if closed, by default True

Returns:

Database connection

Return type:

sqlalchemy.engine.base.Connection

Raises:

ValueError – If connection doesn’t exist and open_if_not_exist is False

create_all(bind: _CreateDropBind, tables: _typing_Sequence[Table] | None = None, checkfirst: bool = True) → None

Create all tables stored in this metadata.

Conditional by default, will not attempt to recreate tables already present in the target database.

Parameters:

  • bind – A Connection or Engine used to access the database.

  • tables – Optional list of Table objects, which is a subset of the total tables in the MetaData (others are ignored).

  • checkfirst – Defaults to True, don’t issue CREATEs for tables already present in the target database.

create_drop_stringify_dialect = 'default'
property dialect: Dialect

Get the SQLAlchemy dialect for the bound engine.

Returns:

Database dialect

Return type:

sqlalchemy.engine.interfaces.Dialect

property dialect_name: str

Get the name of the database dialect.

Returns:

Dialect name (e.g., ‘mssql’, ‘oracle’, ‘sqlite’)

Return type:

str

dispatch = <sqlalchemy.event.base.DDLEventsDispatch object>
dispose()[source]

This method leaves the possibility of checked-out connections remaining open, as it only affects connections that are idle in the pool.

drop_all(bind: _CreateDropBind, tables: _typing_Sequence[Table] | None = None, checkfirst: bool = True) → None

Drop all tables stored in this metadata.

Conditional by default, will not attempt to drop tables not present in the target database.

Parameters:

  • bind – A Connection or Engine used to access the database.

  • tables – Optional list of Table objects, which is a subset of the total tables in the MetaData (others are ignored).

  • checkfirst – Defaults to True, only issue DROPs for tables confirmed to be present in the target database.

drop_table_if_exists(table_name: str, schema: str | None = None, connection_name: str | None = None, transaction: bool = False, auto_close: bool = False)[source]

Drop a table if it exists using database-specific syntax.

Parameters:

  • table_name (str) – Name of the table to drop

  • schema (str, optional) – Schema containing the table

  • connection_name (str, optional) – Name of the connection to use

  • transaction (bool, optional) – Whether to use a transaction, by default False

  • auto_close (bool, optional) – Whether to close connection after execution, by default False

execute(sql: str | Executable, *list_params, transaction: bool = True, auto_close: bool = True, connection_name: str | None = None, **params) → Result | CursorResult[source]

Execute SQL statement with optional transaction management.

Parameters:

  • sql (str or sqlalchemy.sql.expression.Executable) – SQL statement to execute

  • *list_params (tuple) – Positional parameters for the SQL

  • transaction (bool, optional) – Whether to wrap in a transaction, by default True

  • auto_close (bool, optional) – Whether to close connection after execution, by default True

  • connection_name (str, optional) – Name of the connection to use

  • **params (dict) – Named parameters for the SQL

Returns:

Query result

Return type:

sqlalchemy.engine.Result

execute_direct(sql: str, return_results: bool = False) → Sequence[Row] | None[source]

Execute SQL directly using raw DBAPI connection.

Parameters:

  • sql (str) – SQL statement to execute

  • return_results (bool, optional) – Whether to return result rows, by default False

Returns:

List of result rows if return_results is True, otherwise None

Return type:

list or None

execute_inside_trans(sql: str | Executable, *list_params, connection_name: str | None = None, **params) → Result | CursorResult[source]

Execute SQL within an existing transaction. Do not open a new transaction. Do not close the connection.

Parameters:
execute_procedure(procedure_name: str, *args, return_results: bool = False, dpapi_connection=None) → Sequence[Row] | None[source]

Execute a stored procedure

Parameters:

  • procedure_name (str) – The procedure to run.

  • args – The arguments to pass

  • return_results – Needs to be a keyword param. Should we try and get result rows from the procedure.

  • dpapi_connection – A raw dpapi connection to use. Optional.

Raises:

  • sqlalchemy.exc.DBAPIError: – API error

  • sqlalchemy.exc.DatabaseError: – Proxy for database error

has_active_transaction(connection_name: str | None = None) → bool[source]

Check if a connection has an active transaction.

Parameters:

connection_name (str, optional) – Name of the connection to check

Returns:

True if an active transaction exists

Return type:

bool

info

Info dictionary associated with the object, allowing user-defined data to be associated with this SchemaItem.

The dictionary is automatically generated when first accessed. It can also be specified in the constructor of some objects, such as _schema.Table and _schema.Column.

is_connected(connection_name: str | None = None) → bool[source]

Check if a connection exists and is open.

Parameters:

connection_name (str, optional) – Name of the connection to check

Returns:

True if connection exists and is open

Return type:

bool

static qualified_name(schema: str | None, table: str) → str[source]

Create a qualified table name from schema and table.

Parameters:

  • schema (str or None) – Schema name

  • table (str) – Table name

Returns:

Qualified table name (schema.table) or just table name if no schema

Return type:

str

reflect(bind: Engine | Connection, schema: str | None = None, views: bool = False, only: _typing_Sequence[str] | Callable[[str, MetaData], bool] | None = None, extend_existing: bool = False, autoload_replace: bool = True, resolve_fks: bool = True, **dialect_kwargs: Any) → None

Load all available table definitions from the database.

Automatically creates Table entries in this MetaData for any table available in the database but not yet present in the MetaData. May be called multiple times to pick up tables recently added to the database, however no special action is taken if a table in this MetaData no longer exists in the database.

Parameters:

  • bind – A Connection or Engine used to access the database.

  • schema – Optional, query and reflect tables from an alternate schema. If None, the schema associated with this _schema.MetaData is used, if any.

  • views – If True, also reflect views (materialized and plain).

  • only –

    Optional. Load only a sub-set of available named tables. May be specified as a sequence of names or a callable.

    If a sequence of names is provided, only those tables will be reflected. An error is raised if a table is requested but not available. Named tables already present in this MetaData are ignored.

    If a callable is provided, it will be used as a boolean predicate to filter the list of potential table names. The callable is called with a table name and this MetaData instance as positional arguments and should return a true value for any table to reflect.

  • extend_existing – Passed along to each _schema.Table as _schema.Table.extend_existing.

  • autoload_replace – Passed along to each _schema.Table as _schema.Table.autoload_replace.

  • resolve_fks –

    if True, reflect _schema.Table objects linked to _schema.ForeignKey objects located in each _schema.Table. For _schema.MetaData.reflect(), this has the effect of reflecting related tables that might otherwise not be in the list of tables being reflected, for example if the referenced table is in a different schema or is omitted via the MetaData.reflect.only parameter. When False, _schema.ForeignKey objects are not followed to the _schema.Table in which they link, however if the related table is also part of the list of tables that would be reflected in any case, the _schema.ForeignKey object will still resolve to its related _schema.Table after the _schema.MetaData.reflect() operation is complete. Defaults to True.

    Added in version 1.3.0.

    See also

    _schema.Table.resolve_fks

  • **dialect_kwargs – Additional keyword arguments not mentioned above are dialect specific, and passed in the form <dialectname>_<argname>. See the documentation regarding an individual dialect at Dialects for detail on documented arguments.

See also

Reflecting Database Objects

_events.DDLEvents.column_reflect() - Event used to customize the reflected columns. Usually used to generalize the types using _types.TypeEngine.as_generic()

Reflecting with Database-Agnostic Types - describes how to reflect tables using general types.

remove(table: Table) → None

Remove the given Table object from this MetaData.

rename_table(schema: str | None, table_name: str, new_table_name: str)[source]

Rename a table using database-specific syntax.

Parameters:

  • schema (str or None) – Schema containing the table

  • table_name (str) – Current table name

  • new_table_name (str) – New table name

resolve_connection_name(connection_name: str | None = None) → str[source]

Resolve connection name to use, applying database-specific rules and defaults.

Parameters:

connection_name (str, optional) – Connection name to resolve, uses default if None

Returns:

Resolved connection name

Return type:

str

rollback(connection_name: str | None = None)[source]

Rollback the active transaction for a connection.

Parameters:

connection_name (str, optional) – Name of the connection

Raises:

RuntimeError – If no active transaction exists

schema
session() → Session[source]

Create a new SQLAlchemy ORM session.

Returns:

New database session

Return type:

sqlalchemy.orm.Session

property sorted_tables: List[Table]

Returns a list of _schema.Table objects sorted in order of foreign key dependency.

The sorting will place _schema.Table objects that have dependencies first, before the dependencies themselves, representing the order in which they can be created. To get the order in which the tables would be dropped, use the reversed() Python built-in.

Warning

The MetaData.sorted_tables attribute cannot by itself accommodate automatic resolution of dependency cycles between tables, which are usually caused by mutually dependent foreign key constraints. When these cycles are detected, the foreign keys of these tables are omitted from consideration in the sort. A warning is emitted when this condition occurs, which will be an exception raise in a future release. Tables which are not part of the cycle will still be returned in dependency order.

To resolve these cycles, the _schema.ForeignKeyConstraint.use_alter parameter may be applied to those constraints which create a cycle. Alternatively, the _schema.sort_tables_and_constraints() function will automatically return foreign key constraints in a separate collection when cycles are detected so that they may be applied to a schema separately.

Changed in version 1.3.17: - a warning is emitted when MetaData.sorted_tables cannot perform a proper sort due to cyclical dependencies. This will be an exception in a future release. Additionally, the sort will continue to return other tables not involved in the cycle in dependency order which was not the case previously.

See also

_schema.sort_tables()

_schema.sort_tables_and_constraints()

_schema.MetaData.tables

_reflection.Inspector.get_table_names()

_reflection.Inspector.get_sorted_table_and_fkc_names()

table_inventory(schema: str | None = None, force_reload: bool = False) → set[str][source]

Get a case-insensitive set of table names for a schema.

Parameters:

  • schema (str, optional) – Schema name to query, None for default schema

  • force_reload (bool, optional) – Whether to force reload from database, by default False

Returns:

Set of table names in the schema

Return type:

CaseInsensitiveSet

tables

A dictionary of _schema.Table objects keyed to their name or “table key”.

The exact key is that determined by the _schema.Table.key attribute; for a table with no _schema.Table.schema attribute, this is the same as _schema.Table.name. For a table with a schema, it is typically of the form schemaname.tablename.

See also

_schema.MetaData.sorted_tables

transaction(connection_name: str | None = None) → Transaction | None[source]

Get the current transaction for a connection or None if no transaction has been started.

Parameters:

connection_name (str, optional) – Name of the connection

Returns:

Current transaction, or None if no active transaction

Return type:

sqlalchemy.engine.base.Transaction

property uses_bytes_length_limits: bool

Check if database uses byte-based rather than character-based length limits.

Returns:

True if database uses byte-based length limits (e.g., Oracle, Redshift)

Return type:

bool