ds_provider_postgresql_py_lib.utils¶

File: __init__.py Region: ds_provider_postgresql_py_lib/utils

Utils Package

This package exports shared helper functions used across provider modules.

Submodules¶

Functions¶

`validate_duplicate_identity_rows`(→ None)	Ensure input does not contain duplicate identity values.
`validate_identity_columns`(→ None)	Ensure identity columns exist in both input and target table.
`execute_returning_rows`(→ list[dict[str, Any]])	Execute a statement with RETURNING and convert rows to dicts.
`output_from_rows`(→ pandas.DataFrame)	Build output DataFrame using a stable table-column order.
`pandas_dtype_to_sqlalchemy`(→ dict[str, Any])	Convert pandas dtypes Series to a dict mapping column names to SQLAlchemy types.
`quote_identifier`(→ str)	Safely quote SQL identifiers for dynamic DDL statements.

Package Contents¶

ds_provider_postgresql_py_lib.utils.validate_duplicate_identity_rows(content: pandas.DataFrame, identity_columns: collections.abc.Sequence[str]) → None[source]¶

Ensure input does not contain duplicate identity values.

Parameters:

content – Input DataFrame.
identity_columns – Identity columns used for matching.

Returns:

None

Raises:

ValidationError – If duplicate identity rows are found.

ds_provider_postgresql_py_lib.utils.validate_identity_columns(table: sqlalchemy.Table, identity_columns: collections.abc.Sequence[str], content: pandas.DataFrame) → None[source]¶

Ensure identity columns exist in both input and target table.

Parameters:

table – Reflected SQLAlchemy table.
identity_columns – Identity columns used for matching.
content – Input DataFrame.

Returns:

None

Raises:

ValidationError – If identity configuration is invalid.

ds_provider_postgresql_py_lib.utils.execute_returning_rows(conn: sqlalchemy.engine.Connection, stmt: sqlalchemy.sql.Executable) → list[dict[str, Any]][source]¶

Execute a statement with RETURNING and convert rows to dicts.

Parameters:

conn – SQLAlchemy connection.
stmt – SQL statement to execute.

Returns:

Returned rows as dictionaries.

Return type:

list[dict[str, Any]]

ds_provider_postgresql_py_lib.utils.output_from_rows(table: sqlalchemy.Table, rows: list[dict[str, Any]]) → pandas.DataFrame[source]¶

Build output DataFrame using a stable table-column order.

Parameters:

table – Reflected SQLAlchemy table.
rows – Row dictionaries to materialize.

Returns:

Output DataFrame aligned to table column order.

Return type:

pd.DataFrame

ds_provider_postgresql_py_lib.utils.pandas_dtype_to_sqlalchemy(dtypes: pandas.Series) → dict[str, Any][source]¶

Convert pandas dtypes Series to a dict mapping column names to SQLAlchemy types.

Parameters:: dtypes – Pandas dtype series keyed by column name.
Returns:: SQLAlchemy column types keyed by column name.
Return type:: dict[str, Any]

ds_provider_postgresql_py_lib.utils.quote_identifier(identifier: str) → str[source]¶

Safely quote SQL identifiers for dynamic DDL statements.

Parameters:: identifier – Raw SQL identifier.
Returns:: Properly quoted SQL identifier.
Return type:: str