operations¶

Advanced database operations for SQLAlchemy.

This module provides high-performance database operations that extend beyond basic CRUD functionality. It implements specialized database operations optimized for bulk data handling and schema management.

The operations module is designed to work seamlessly with SQLAlchemy Core and ORM, providing efficient implementations for common database operations patterns.

Features¶

Cross-database ON CONFLICT/ON DUPLICATE KEY UPDATE operations
MERGE statement support for Oracle and PostgreSQL 15+

Security¶

This module constructs SQL statements using database identifiers (table and column names) that MUST come from trusted sources only. All identifiers should originate from:

SQLAlchemy model metadata (e.g., Model.__table__)
Hardcoded strings in application code
Validated configuration files

Never pass user input directly as table names, column names, or other SQL identifiers. Data values are properly parameterized using bindparam() to prevent SQL injection.

Notes:¶

This module is designed to be database-agnostic where possible, with specialized optimizations for specific database backends where appropriate.

See Also:¶

sqlalchemy.sql.expression : SQLAlchemy Core expression language
sqlalchemy.orm : SQLAlchemy ORM functionality
advanced_alchemy.extensions : Additional database extensions

class advanced_alchemy.operations.MergeStatement[source]¶

Bases: Executable, ClauseElement

A MERGE statement for Oracle and PostgreSQL 15+.

This provides a high-level interface for MERGE operations that can handle both matched and unmatched conditions.

inherit_cache = True¶

Indicate if this HasCacheKey instance should make use of the cache key generation scheme used by its immediate superclass.

The attribute defaults to None, which indicates that a construct has not yet taken into account whether or not its appropriate for it to participate in caching; this is functionally equivalent to setting the value to False, except that a warning is also emitted.

This flag can be set to True on a particular class, if the SQL that corresponds to the object does not change based on attributes which are local to this class, and not its superclass.

See also

Enabling Caching Support for Custom Constructs - General guideslines for setting the HasCacheKey.inherit_cache attribute for third-party or user defined SQL constructs.

__init__(table, source, on_condition, when_matched_update=None, when_not_matched_insert=None)[source]¶

Initialize a MERGE statement.

Parameters:

table¶ (Table) – Target table for the merge operation
source¶ (Union[ClauseElement, str]) – Source data (can be a subquery or table)
on_condition¶ (ClauseElement) – Condition for matching rows
when_matched_update¶ (Optional[dict[str, Any]]) – Values to update when rows match
when_not_matched_insert¶ (Optional[dict[str, Any]]) – Values to insert when rows don’t match
table (Table)
source (ClauseElement | str)
on_condition (ClauseElement)
when_matched_update (dict[str, Any] | None)
when_not_matched_insert (dict[str, Any] | None)

Return type:

None

class advanced_alchemy.operations.OnConflictUpsert[source]¶

Bases: object

Cross-database upsert operation using dialect-specific constructs.

This class provides a unified interface for upsert operations across different database backends using their native ON CONFLICT or ON DUPLICATE KEY UPDATE mechanisms.

static supports_native_upsert(dialect_name)[source]¶

Check if the dialect supports the single-row create_upsert API.

This flag is scoped to the per-row INSERT ... ON CONFLICT / ON DUPLICATE KEY UPDATE dialects that OnConflictUpsert.create_upsert() can compile directly. The bulk MERGE (mssql, oracle) and INSERT OR UPDATE (spanner) primitives are dispatched separately by Repository.upsert_many() via resolve_upsert_strategy() and are intentionally not reported here.

Parameters:

dialect_name¶ (str) – Name of the database dialect
dialect_name (str)

Return type:

bool

Returns:

True for postgresql / cockroachdb / sqlite / mysql / mariadb / duckdb; False otherwise.

static create_upsert(table, values, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False)[source]¶

Create a dialect-specific upsert statement.

Parameters:

table¶ (Table) – Target table for the upsert
values¶ (dict[str, Any]) – Values to insert/update
conflict_columns¶ (list[str]) – Columns that define the conflict condition
update_columns¶ (Optional[list[str]]) – Columns to update on conflict (defaults to all non-conflict columns)
dialect_name¶ (Optional[str]) – Database dialect name (auto-detected if not provided)
validate_identifiers¶ (bool) – If True, validate column names for safety (default: False)
table (Table)
values (dict[str, Any])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)

Return type:

Insert

Returns:

A SQLAlchemy Insert for the ON-CONFLICT dialects (postgresql / cockroachdb / sqlite / duckdb / mysql / mariadb). MSSQL, Oracle, and Spanner are handled by the bulk MERGE / INSERT OR UPDATE path in OnConflictUpsert.create_merge_many() and SpannerUpsert, accessed through resolve_upsert_strategy() from the repository layer — they are not exposed via this single-row API.

Raises:

NotImplementedError – If the dialect doesn’t support native upsert
ValueError – If validate_identifiers is True and invalid identifiers are found

static create_merge_upsert(table, values, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False)[source]¶

Create a MERGE-based upsert for Oracle/PostgreSQL 15+.

For Oracle databases, this method automatically generates values for primary key columns that have callable defaults (such as UUID generation functions). This is necessary because Oracle MERGE statements cannot use Python callable defaults directly in the INSERT clause.

Parameters:

table¶ (Table) – Target table for the upsert
values¶ (dict[str, Any]) – Values to insert/update
conflict_columns¶ (list[str]) – Columns that define the matching condition
update_columns¶ (Optional[list[str]]) – Columns to update on match (defaults to all non-conflict columns)
dialect_name¶ (Optional[str]) – Database dialect name (used to determine Oracle-specific syntax)
validate_identifiers¶ (bool) – If True, validate column names for safety (default: False)
table (Table)
values (dict[str, Any])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)

Return type:

tuple[MergeStatement, dict[str, Any]]

Returns:

A tuple of (MergeStatement, additional_params) where additional_params contains any generated values (like Oracle UUID primary keys)

Raises:

ValueError – If validate_identifiers is True and invalid identifiers are found

static create_upsert_many(table, values_list, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False)[source]¶

Build a dialect-specific bulk Insert with ON CONFLICT / ON DUPLICATE KEY UPDATE.

Compiles to a single INSERT ... VALUES (...), (...), ... per chunk so the round-trip cost is fixed regardless of batch size.

Parameters:

table¶ (Table) – Target table for the upsert.
values_list¶ (list[dict[str, Any]]) – Rows to insert/update. All rows MUST share the same keys.
conflict_columns¶ (list[str]) – Columns that define the conflict / match condition.
update_columns¶ (Optional[list[str]]) – Columns to update on conflict (defaults to all non-conflict keys from the first row).
dialect_name¶ (Optional[str]) – Database dialect name; determines compile path.
validate_identifiers¶ (bool) – If True, validate column identifiers for safety.
table (Table)
values_list (list[dict[str, Any]])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)

Return type:

tuple[Insert, bool]

Returns:

A tuple (statement, supports_returning) where supports_returning is True for postgresql / cockroachdb / sqlite / duckdb and False for mysql / mariadb.

Raises:

ValueError – values_list is empty, rows have heterogeneous keys, or identifier validation fails.
NotImplementedError – The dialect does not support an ON CONFLICT style native bulk upsert.

static create_merge_many(table, values_list, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False)[source]¶

Build a bulk MERGE / executemany-fallback per dialect.

Returns a single MergeStatement for dialects whose MERGE syntax supports a multi-row source (oracle, mssql, postgresql/cockroachdb), and a list of single-row MergeStatement (one per input row) for everything else.

Parameters:

table¶ (Table) – Target table for the upsert.
values_list¶ (list[dict[str, Any]]) – Rows to insert/update. All rows MUST share the same keys.
conflict_columns¶ (list[str]) – Columns that define the matching condition.
update_columns¶ (Optional[list[str]]) – Columns to update on match (defaults to all non-conflict keys from the first row).
dialect_name¶ (Optional[str]) – Database dialect name; selects the source construction.
validate_identifiers¶ (bool) – If True, validate column identifiers for safety.
table (Table)
values_list (list[dict[str, Any]])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)

Return type:

tuple[Union[MergeStatement, list[MergeStatement]], dict[str, Any]]

Returns:

A tuple (statement_or_list, additional_params). additional_params carries generated values (Oracle UUID PKs, MSSQL bound row values) that must be passed when executing.

Raises:

ValueError – values_list is empty, rows have heterogeneous keys, or identifier validation fails.

class advanced_alchemy.operations.SpannerUpsert[source]¶

Bases: Executable, ClauseElement

Spanner-specific bulk upsert primitive (INSERT OR UPDATE INTO).

Cloud Spanner does not implement SQL MERGE; the closest DML form is INSERT OR UPDATE INTO {table} ({cols}) VALUES (...), (...). We model this with its own ClauseElement (instead of overloading MergeStatement) because the syntax has no USING / WHEN MATCHED shape.

The PK must be present in every row — Spanner does not auto-generate PKs via DML. INSERT OR UPDATE has no RETURNING/OUTPUT equivalent; callers needing hydration must re-SELECT.

inherit_cache = True¶

Indicate if this HasCacheKey instance should make use of the cache key generation scheme used by its immediate superclass.

This flag can be set to True on a particular class, if the SQL that corresponds to the object does not change based on attributes which are local to this class, and not its superclass.

See also

Enabling Caching Support for Custom Constructs - General guideslines for setting the HasCacheKey.inherit_cache attribute for third-party or user defined SQL constructs.

__init__(table, values_list)[source]¶

Initialize a Spanner INSERT_OR_UPDATE.

Parameters:

table¶ (Table) – Target table for the upsert.
values_list¶ (list[dict[str, Any]]) – Rows to insert/update. All rows MUST share the same keys; PK columns MUST be present in every row.
table (Table)
values_list (list[dict[str, Any]])

Raises:

ValueError – values_list is empty or rows have heterogeneous keys.

Return type:

None

columns: tuple[str, ...]¶

class advanced_alchemy.operations.UpsertStrategy[source]¶

Bases: NamedTuple

Dispatch decision returned by resolve_upsert_strategy().

Tells the repository which native primitive to compile (on_conflict / merge / insert_or_update) or whether to take the existing SELECT-then-partition fallback. The conflict_columns field is the validated unique key (PK / UniqueConstraint / unique Index) — which may be a subset of the caller’s match_fields (e.g. PK match where the caller passed extra columns).

kind: Literal['on_conflict', 'merge', 'insert_or_update', 'fallback']¶: Alias for field number 0

supports_returning: bool¶: Alias for field number 1

conflict_columns: tuple[str, ...]¶: Alias for field number 2

dialect_name: str¶: Alias for field number 3

static __new__(_cls, kind: Literal['on_conflict', 'merge', 'insert_or_update', 'fallback'], supports_returning: bool, conflict_columns: tuple[str, ...], dialect_name: str)¶

Create new instance of UpsertStrategy(kind, supports_returning, conflict_columns, dialect_name)

Parameters:

kind (Literal['on_conflict', 'merge', 'insert_or_update', 'fallback'])
supports_returning (bool)
conflict_columns (tuple[str, ...])
dialect_name (str)

advanced_alchemy.operations.compile_spanner_upsert_default(element, compiler, **kwargs)[source]¶

Default compilation - raises error for non-spanner dialects.

Return type:

str

Parameters:

element (SpannerUpsert)
compiler (SQLCompiler)
kwargs (Any)

advanced_alchemy.operations.resolve_upsert_strategy(table, match_fields, dialect_name)[source]¶

Resolve the optimal upsert strategy for (table, match_fields, dialect).

The result is cached for the process lifetime; Table objects are singletons per declarative class, so this is one decision per (model, match_fields, dialect) tuple.

Resolution priority:

match_fields equals or is a superset of the table’s primary key → native primitive for the dialect, conflict_columns is the PK.
A UniqueConstraint whose columns match exactly → native primitive, conflict_columns is that constraint’s columns.
A unique Index whose columns match exactly → native primitive, conflict_columns is that index’s columns.
Otherwise → kind="fallback", supports_returning=False.

Parameters:

table¶ (Table) – Target table. Used by identity for caching.
match_fields¶ (Sequence[str]) – Columns the caller wants to match on. Order-insensitive.
dialect_name¶ (str) – Database dialect name.
table (Table)
match_fields (Sequence[str])
dialect_name (str)

Return type:

UpsertStrategy

Returns:

An UpsertStrategy describing the decision.

Raises:

ValueError – match_fields is empty or contains a column not present on the table.

advanced_alchemy.operations.validate_identifier(name, identifier_type='identifier')[source]¶

Validate a SQL identifier to ensure it’s safe for use in SQL statements.

This function provides validation for SQL identifiers (table names, column names, etc.) to ensure they contain only safe characters. While the operations in this module should only receive identifiers from trusted sources, this validation adds an extra layer of security.

Note: SQL keywords (like ‘select’, ‘insert’, etc.) are allowed as they can be properly quoted/escaped by SQLAlchemy when used as identifiers.

Parameters:

name¶ (str) – The identifier to validate
identifier_type¶ (str) – Type of identifier for error messages (e.g., “column”, “table”)
name (str)
identifier_type (str)

Return type:

str

Returns:

The validated identifier

Raises:

ValueError – If the identifier is empty or contains invalid characters

Examples

>>> validate_identifier("user_id")
'user_id'
>>> validate_identifier("users_table", "table")
'users_table'
>>> validate_identifier("select")  # SQL keywords are allowed
'select'
>>> validate_identifier(
...     "drop table users; --"
... )  # Raises ValueError - contains invalid characters