utils

sqlalchemy_upsert_kit.utils.get_pk_name(table: Table) → str

Extract the primary key column name from a SQLAlchemy table.

This function ensures the table has exactly one primary key column, which is required for the bulk operations to work correctly.

Args:

table: SQLAlchemy table object

Returns:

Name of the primary key column

Raises:

ValueError: If table has zero or multiple primary key columns

Example:

>>> table = sa.Table(
...     'users',
...     metadata,
...     sa.Column('id', sa.Integer, primary_key=True)
... )
>>> get_pk_name(table)
'id'

sqlalchemy_upsert_kit.utils.clone_temp_table(original_table: Table, metadata: MetaData, temp_table_name: str | None = None) → Table

Create a temporary table with the same schema as the original table.

This function clones the structure of an existing table to create a temporary table for bulk operations. The temporary table inherits all columns, types, and constraints from the original table.

Args:

original_table: The table to clone metadata: Metadata object for the temporary table.

Should be a separate instance from the original table’s metadata to avoid conflicts and enable proper cleanup.

temp_table_name: Custom name for the temporary table.

If None, generates a unique name with timestamp to avoid conflicts in concurrent environments.

Returns:

New temporary table with identical schema

Note

• Use a separate MetaData instance to isolate the temporary table

• In high-concurrency scenarios, consider providing unique temp_table_name

• The temporary table is not automatically bound to any engine

Example:

>>> metadata = sa.MetaData()
>>> _temp_table = clone_temp_table(users_table, metadata)
>>> # _temp_table has same columns as users_table but different metadata