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: Optional[bool] = 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 (sqlalchemy.schema.Table) – Target table for the merge operation

  • source (ClauseElement | str) – Source data (can be a subquery or table)

  • on_condition (ClauseElement) – Condition for matching rows

  • when_matched_update (dict[str, typing.Any] | None) – Values to update when rows match

  • when_not_matched_insert (dict[str, typing.Any] | None) – Values to insert when rows don’t match

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 native upsert operations.

Parameters:

dialect_name (str) – Name of the database dialect

Return type:

bool

Returns:

True if native upsert is supported, 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 (sqlalchemy.schema.Table) – Target table for the upsert

  • values (dict[str, typing.Any]) – Values to insert/update

  • conflict_columns (list[str]) – Columns that define the conflict condition

  • update_columns (list[str] | None) – Columns to update on conflict (defaults to all non-conflict columns)

  • dialect_name (str | None) – Database dialect name (auto-detected if not provided)

  • validate_identifiers (bool) – If True, validate column names for safety (default: False)

Return type:

Insert

Returns:

A SQLAlchemy Insert statement with upsert logic

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 (sqlalchemy.schema.Table) – Target table for the upsert

  • values (dict[str, typing.Any]) – Values to insert/update

  • conflict_columns (list[str]) – Columns that define the matching condition

  • update_columns (list[str] | None) – Columns to update on match (defaults to all non-conflict columns)

  • dialect_name (str | None) – Database dialect name (used to determine Oracle-specific syntax)

  • validate_identifiers (bool) – If True, validate column names for safety (default: False)

Return type:

tuple[MergeStatement, dict[str, typing.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

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”)

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