config_wrangler.config_templates.sqlalchemy_database module

pydantic model config_wrangler.config_templates.sqlalchemy_database.SQLAlchemyDatabase[source]

Bases: Credentials

Configuration for SQLAlchemy database connections.

Extends Credentials to provide SQLAlchemy-specific connection management including support for multiple dialects, connection pooling, and AWS Redshift temporary credentials.

Parameters:

  • dialect (str) – The SQLAlchemy dialect to use (e.g., ‘postgresql’, ‘mysql’, ‘oracle’, ‘sqlite’). See https://docs.sqlalchemy.org/en/20/dialects/

  • driver (str, optional) – The Python database driver to use (e.g., ‘psycopg2’ for PostgreSQL).

  • host (str, optional) – Hostname or IP address of the database server. Required for non-sqlite databases.

  • port (int, optional) – Port number for the database connection.

  • database_name (str) – Name of the database to connect to.

  • use_get_cluster_credentials (bool, default False) – If True, use AWS Redshift temporary credentials via boto3.

  • rs_db_user_id (str, optional) – Redshift database user ID for temporary credential requests.

  • create_engine_args (dict, optional) – Additional arguments to pass to SQLAlchemy’s create_engine function.

  • dbapi_args (dict, optional) – Additional arguments to pass to the DBAPI via URI query parameters.

See also

config_wrangler.config_templates.credentials.Credentials

Base class for credential management

Config:

  • validate_default: bool = True

  • validate_assignment: bool = True

  • validate_credentials: bool = True

Fields:

  • arraysize (int | None)

  • aws_access_key_id (str | None)

  • aws_secret_access_key (str | None)

  • create_engine_args (Dict[str, Any])

  • database_name (str)

  • dbapi_args (Dict[str, Any])

  • dialect (str)

  • driver (str | None)

  • encoding (str | None)

  • host (str | None)

  • keepass ()

  • keepass_config ()

  • keepass_group ()

  • keepass_title ()

  • keyring_section ()

  • password_source ()

  • poolclass (str | None)

  • port (int | None)

  • raw_password ()

  • rs_auto_create (bool | None)

  • rs_cluster_id (str | None)

  • rs_db_groups (List[str] | None)

  • rs_db_user_id (str | None)

  • rs_duration_seconds (int | None)

  • rs_new_credentials_seconds (int)

  • rs_region_name (str | None)

  • use_get_cluster_credentials (bool)

  • user_id ()

  • validate_password_on_load ()

Validators:
add_child(name: str, child_object: ConfigHierarchy)

Set this configuration as a child in the hierarchy of another config. For any programmatically created config objects this is required so that the new object ‘knows’ where it lives in the hierarchy – most importantly so that it can find the hierarchies root object.

validator check_model  »  all fields[source]
connect() → sqlalchemy.engine.base.Connection[source]

Connect to the configured database.

Creates a new SQLAlchemy Connection object. For SQLite databases, the same connection is reused since it does not support concurrent connections.

Returns:

A SQLAlchemy Connection object for executing SQL statements.

Return type:

sqlalchemy.engine.base.Connection

Notes

SQLite connections are cached and reused. For other databases, a new connection is obtained from the connection pool on each call.

See also

get_engine

Creates or retrieves the SQLAlchemy engine

session

Creates a SQLAlchemy session for ORM operations

full_item_name(item_name: str | None = None, delimiter: str = ' -> ')

The fully qualified name of this config item in the config hierarchy.

get_cluster_credentials() → Tuple[str, str][source]

Retrieve temporary database credentials from AWS Redshift.

Makes a boto3 call to get_cluster_credentials to obtain temporary database credentials. Manages credential expiration and refreshes them as needed based on the configured duration settings.

Returns:

A tuple containing (username, password) for the database connection.

Return type:

tuple of (str, str)

Raises:

  • ValueError – If required Redshift configuration parameters are missing.

  • SQLAlchemyError – If the AWS response is invalid or missing required fields.

Notes

Credential expiry is managed automatically based on the minimum of: - The configured rs_new_credentials_seconds duration - The server-provided expiration time minus a 15-minute safety margin - IAM session expiry (if using AWS_ASSUME_ROLE)

get_copy(copied_by: str = 'get_copy') → ConfigHierarchy

Copy this configuration. Useful when you need to programmatically modify a configuration without modifying the original base configuration.

get_engine() → Engine[source]

Get or create the SQLAlchemy engine for this database connection.

Creates a new engine on first call and caches it for reuse. For Redshift with temporary credentials enabled, sets up event listeners to handle credential refresh and connection validation.

Returns:

The SQLAlchemy engine instance for this database.

Return type:

sqlalchemy.engine.Engine

Raises:

ValueError – If an invalid poolclass value is configured.

Notes

The engine is configured with the following optional parameters: - arraysize (Oracle only) - encoding - poolclass (QueuePool or NullPool) - Additional arguments from create_engine_args

For Redshift temporary credentials, event listeners are registered to: - Refresh credentials on connection (do_connect) - Validate credentials on checkout - Perform pessimistic connection testing (engine_connect) - Handle IAM authentication token expiration errors

See also

get_cluster_credentials

Retrieves temporary Redshift credentials (called automatically by this method)

get_password() → str[source]

Get the password for this resource. password_source controls where it looks for the password. If that is None, then the root level passwords container is checked for password_source value.

get_uri() → URL[source]

Construct a SQLAlchemy database URI from the configuration.

Builds a complete connection URI including credentials, host, port, and database name. For Redshift with use_get_cluster_credentials enabled, retrieves temporary credentials via get_cluster_credentials().

Returns:

A SQLAlchemy URL object representing the database connection string.

Return type:

sqlalchemy.engine.url.URL

Raises:

ValueError – If rs_db_user_id is missing when use_get_cluster_credentials is True.

See also

get_cluster_credentials

Retrieves temporary Redshift credentials (called automatically by this method)

model_dump_non_private(*, mode: Literal['json', 'python'] | str = 'python', exclude: Set[str] | None = None) → dict[str, Any]
raw_connection()[source]
Returns:

A raw DBAPI connection object from the underlying connection pool.

Return type:

connection

See also

get_engine

Creates or retrieves the SQLAlchemy engine

connect

Returns a SQLAlchemy Connection wrapper

Notes

See https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Engine.raw_connection

session() → Session[source]

Creates a new SQLAlchemy Session bound to this database’s engine for performing ORM operations.

Returns:

A SQLAlchemy Session object for ORM-based database operations.

Return type:

sqlalchemy.orm.Session

See also

get_engine

Creates or retrieves the SQLAlchemy engine

connect

Creates a connection for core SQL operations

set_as_child(name: str, other_config_item: ConfigHierarchy)
validator translate  »  all fields[source]
static translate_config_data(config_data: MutableMapping)

Children classes can provide translation logic to allow older config files to be used with newer config class definitions.