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, ON DUPLICATE KEY UPDATE, MERGE, or INSERT OR 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 OnConflictUpsert.create_insert_or_update_many(), accessed through resolve_upsert_strategy() from the repository layer.

Raises:

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

static create_insert_or_update_many(table, rows)[source]#

Create a Spanner INSERT OR UPDATE using SQLAlchemy’s Insert.

A regular Insert retains result-column metadata, allowing callers to add returning(model_type). The Spanner dialect compiles that combination to THEN RETURN.

Parameters:

table¶ (Table) – Target table for the upsert.
rows¶ (Sequence[dict[str, typing.Any]]) – Homogeneous rows to insert or update.
table (Table)
rows (Sequence[dict[str, TypeAliasForwardRef('typing.Any')]])

Return type:

Insert

Returns:

A multi-values insert prefixed with Spanner’s OR UPDATE token.

Raises:

ValueError – rows is empty or rows have heterogeneous keys.

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. Since Oracle requires FROM DUAL for SELECT statements without tables, the Oracle source subquery selects from DUAL.

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, rows, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False, model_type=None)[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.
rows¶ (Sequence[dict[str, typing.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.
model_type¶ (Optional[type[Any]]) – Optional ORM model target used for ORM-aware RETURNING.
table (Table)
rows (Sequence[dict[str, TypeAliasForwardRef('typing.Any')]])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)
model_type (type[Any] | None)

Return type:

Insert

Returns:

A SQLAlchemy Insert statement configured for bulk upsert.

Raises:

ValueError – rows 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, rows, conflict_columns, update_columns=None, dialect_name=None, validate_identifiers=False)[source]#

Build a multi-row MergeStatement for dialects with native MERGE support.

Parameters:

table¶ (Table) – Target table for the upsert.
rows¶ (Sequence[dict[str, typing.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)
rows (Sequence[dict[str, TypeAliasForwardRef('typing.Any')]])
conflict_columns (list[str])
update_columns (list[str] | None)
dialect_name (str | None)
validate_identifiers (bool)

Return type:

MergeStatement

Returns:

A single MergeStatement binding all input rows.

Raises:

ValueError – rows is empty, rows have heterogeneous keys, or identifier validation fails.
NotImplementedError – The dialect does not support bulk MERGE compilation.

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) and has exactly the same columns as the caller’s match_fields.

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

advanced_alchemy.operations.resolve_column_default(arg)[source]#

Invoke a SQLAlchemy ColumnDefault.arg callable, tolerating both signatures.

SQLAlchemy accepts both context-taking defaults (lambda ctx: …) and zero-arg defaults (lambda: …, uuid7, datetime.utcnow). The context-taking shape is attempted first with None; a signature mismatch retries the zero-arg shape. Errors raised inside the default itself — including a context-sensitive default that dereferences the None context — propagate so callers can route the operation through the ORM flush, where a real execution context is available.

Return type:: Any
Parameters:: arg (Any)

advanced_alchemy.operations.resolve_upsert_strategy(table, match_fields, dialect)[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 the table’s primary key → native primitive for the dialect, conflict_columns is the PK.
A UniqueConstraint whose columns match exactly → native primitive where the backend can target it, conflict_columns is that constraint’s columns.
A unique Index whose columns match exactly → native primitive where the backend can target it, conflict_columns is that index’s columns. Spanner INSERT OR UPDATE only matches the primary key, and ambiguous MySQL/MariaDB unique targets fall back.
Otherwise → kind="fallback", supports_returning=False. This includes match_fields entries that do not resolve to table columns (mapped attributes whose name differs from the column key, synonyms, hybrids) — the ORM fallback resolves those via getattr.

Parameters:

table¶ (Table) – Target table. Used by identity for caching.
match_fields¶ (Sequence[str]) – Columns the caller wants to match on. Order-insensitive.
dialect¶ (Union[str, Dialect]) – Database dialect or dialect name. Passing the runtime dialect allows the resolver to honor its actual RETURNING capability.
table (Table)
match_fields (Sequence[str])
dialect (str | Dialect)

Return type:

UpsertStrategy

Returns:

An UpsertStrategy describing the decision.

Raises:

ValueError – match_fields is empty.

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