Skip to main content

databricks.labs.dqx.actions.definition_storage

Storage handlers for persisting DQAction definitions.

This module provides:

  • ActionsStorageHandler — abstract base class with save and load methods.
  • TableActionsStorageHandler — persists actions to a Unity Catalog Delta table via Spark.
  • LakebaseActionsStorageHandler — persists actions to a Lakebase (PostgreSQL) table via SQLAlchemy.
  • ActionsStorageHandlerFactory — selects the right handler from config type.

Schema​

Both handlers store each DQAction as a serialized JSON string alongside the run_config_name and a created_at timestamp:

  • action_json — JSON string produced by ActionSerializer.to_dict.
  • run_config_name — run configuration name for multi-run isolation.
  • created_at — timestamp at write time.

Security​

User-supplied values are sanitized before appearing in log messages (CWE-117).

build_replace_where_predicate​

def build_replace_where_predicate(run_config_name: str) -> str

Build a safe Delta replaceWhere predicate for run_config_name.

Validates that run_config_name contains only characters in the set [A-Za-z0-9_.-] to prevent SQL injection (CWE-89) in the predicate string. If the name contains any other character, UnsafeSqlQueryError is raised before the predicate is constructed.

This mirrors the identical guard in TableChecksStorageHandler.save in checks_storage.py.

Arguments:

  • run_config_name - The run configuration name to embed in the predicate.

Returns:

A SQL predicate string that compares run_config_name against its validated value, safe for use as a Delta replaceWhere option.

Raises:

  • UnsafeSqlQueryError - If run_config_name contains characters outside [A-Za-z0-9_.-].

ActionsStorageHandler Objects​

class ActionsStorageHandler(ABC, Generic[T])

Abstract base class for DQAction definition storage handlers.

Subclasses implement persistence to a specific backend (Delta table or Lakebase PostgreSQL).

Arguments:

  • T - The config type that parameterizes this handler.

save​

@abstractmethod
def save(actions: list[DQAction], config: T) -> None

Persist actions to storage.

Arguments:

  • actions - List of DQAction definitions to persist.
  • config - Backend-specific configuration.

load​

@abstractmethod
def load(config: T) -> list[DQAction]

Load DQAction definitions from storage.

Arguments:

  • config - Backend-specific configuration.

Returns:

List of DQAction instances loaded from storage.

TableActionsStorageHandler Objects​

class TableActionsStorageHandler(
        ActionsStorageHandler[TableActionsStorageConfig])

Persists DQAction definitions to a Unity Catalog Delta table via Spark.

Each DQAction is serialized to JSON via ActionSerializer.to_dict and stored as a single row alongside the run_config_name and a created_at timestamp.

On save, the write mode from config.mode controls whether existing rows for the run_config_name are replaced ("overwrite") or kept ("append").

On load, all rows matching config.run_config_name are read and deserialized.

Arguments:

  • spark - Active SparkSession for Spark-based read/write.
  • ws - Authenticated WorkspaceClient (reserved for future use such as table-existence checks).

save​

def save(actions: list[DQAction], config: TableActionsStorageConfig) -> None

Serialize and write actions to the configured Delta table.

Arguments:

  • actions - List of DQAction definitions to persist.
  • config - TableActionsStorageConfig with target table, mode, and run config name.

Raises:

  • UnsafeSqlQueryError - If run_config_name contains characters outside [A-Za-z0-9_.-] when mode is "overwrite", raised by _build_replace_where to prevent SQL injection in the Delta replaceWhere predicate (CWE-89).

load​

def load(config: TableActionsStorageConfig) -> list[DQAction]

Read and deserialize DQAction definitions from the Delta table.

Returns an empty list when the table does not exist.

Arguments:

  • config - TableActionsStorageConfig with source table and run config name.

Returns:

List of DQAction instances.

LakebaseActionsStorageHandler Objects​

class LakebaseActionsStorageHandler(
        LakebaseConnectionMixin,
        ActionsStorageHandler[LakebaseActionsStorageConfig])

Persists DQAction definitions to a Lakebase (PostgreSQL) table via SQLAlchemy.

Inherits engine lifecycle management from LakebaseConnectionMixin: a lazily created, cached engine with a do_connect listener that refreshes the Databricks-generated credential token before each connection, and schema / table bootstrap on first use.

The handler accepts an optional pre-built Engine for testability — pass an in-memory or test engine via the engine constructor parameter to avoid needing a real Lakebase instance in unit tests.

Arguments:

  • spark - Active SparkSession (kept for interface symmetry; not used for PostgreSQL queries).
  • ws - Authenticated WorkspaceClient used to retrieve the Lakebase DNS and generate short-lived credentials.
  • config - LakebaseActionsStorageConfig with instance and table details.
  • engine - Optional pre-built SQLAlchemy Engine (useful for testing).

save​

def save(actions: list[DQAction],
         config: LakebaseActionsStorageConfig) -> None

Serialize and write actions to the Lakebase table.

Bootstraps the schema and table on first use. When config.mode is "overwrite", all existing rows for config.run_config_name are deleted before inserting the new rows.

Arguments:

  • actions - List of DQAction definitions to persist.
  • config - LakebaseActionsStorageConfig with instance and table details.

load​

def load(config: LakebaseActionsStorageConfig) -> list[DQAction]

Read and deserialize DQAction definitions from the Lakebase table.

Returns an empty list when the table does not exist.

Arguments:

  • config - LakebaseActionsStorageConfig with instance and table details.

Returns:

List of DQAction instances.

ActionsStorageHandlerFactory Objects​

class ActionsStorageHandlerFactory()

Creates the appropriate ActionsStorageHandler for a given config type.

Selection logic:

  • LakebaseActionsStorageConfig → LakebaseActionsStorageHandler
  • TableActionsStorageConfig → TableActionsStorageHandler

Arguments:

  • config - Storage configuration; either TableActionsStorageConfig or LakebaseActionsStorageConfig.
  • spark - Active SparkSession.
  • ws - Authenticated WorkspaceClient.

Returns:

A concrete ActionsStorageHandler instance.

create​

@staticmethod
def create(
    config: TableActionsStorageConfig | LakebaseActionsStorageConfig,
    spark: SparkSession, ws: WorkspaceClient
) -> ActionsStorageHandler[TableActionsStorageConfig] | ActionsStorageHandler[
        LakebaseActionsStorageConfig]

Instantiate the correct storage handler for config.

Arguments:

  • config - Storage configuration.
  • spark - Active SparkSession.
  • ws - Authenticated WorkspaceClient.

Returns:

LakebaseActionsStorageHandler when config is a LakebaseActionsStorageConfig; TableActionsStorageHandler otherwise.

save​

@staticmethod
def save(actions: list[DQAction],
         config: TableActionsStorageConfig | LakebaseActionsStorageConfig,
         spark: SparkSession, ws: WorkspaceClient) -> None

Create the appropriate handler and persist actions to storage.

Arguments:

  • actions - List of DQAction definitions to persist.
  • config - Backend-specific configuration.
  • spark - Active SparkSession.
  • ws - Authenticated WorkspaceClient.

load​

@staticmethod
def load(config: TableActionsStorageConfig | LakebaseActionsStorageConfig,
         spark: SparkSession, ws: WorkspaceClient) -> list[DQAction]

Create the appropriate handler and load DQAction definitions from storage.

Arguments:

  • config - Backend-specific configuration.
  • spark - Active SparkSession.
  • ws - Authenticated WorkspaceClient.

Returns:

List of DQAction instances loaded from storage.