kelp.models.model

Reference for the core model and column configuration models.

kelp.models.model.TableType

Bases: Enum

EXTERNAL class-attribute instance-attribute

EXTERNAL = 'external'

EXTERNAL_SHALLOW_CLONE class-attribute instance-attribute

EXTERNAL_SHALLOW_CLONE = 'external_shallow_clone'

FOREIGN class-attribute instance-attribute

FOREIGN = 'foreign'

MANAGED class-attribute instance-attribute

MANAGED = 'managed'

MANAGED_SHALLOW_CLONE class-attribute instance-attribute

MANAGED_SHALLOW_CLONE = 'managed_shallow_clone'

MATERIALIZED_VIEW class-attribute instance-attribute

MATERIALIZED_VIEW = 'materialized_view'

METRIC_VIEW class-attribute instance-attribute

METRIC_VIEW = 'metric_view'

STREAMING_TABLE class-attribute instance-attribute

STREAMING_TABLE = 'streaming_table'

VIEW class-attribute instance-attribute

VIEW = 'view'

kelp.models.model.Model pydantic-model

Bases: BaseModel

Model definition in Unity Catalog.

Represents a metadata model in Kelp with configuration that maps to Databricks Unity Catalog objects.

Show JSON schema:

{
  "$defs": {
    "AppendConfig": {
      "additionalProperties": false,
      "description": "Append the DataFrame to the target table.",
      "properties": {
        "options": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
          "title": "Options",
          "type": "object"
        },
        "allow_full_refresh": {
          "default": true,
          "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
          "title": "Allow Full Refresh",
          "type": "boolean"
        },
        "mode": {
          "const": "append",
          "default": "append",
          "title": "Mode",
          "type": "string"
        }
      },
      "title": "AppendConfig",
      "type": "object"
    },
    "AutoTTLConfig": {
      "properties": {
        "timestamp_column": {
          "description": "Name of the timestamp column used for TTL",
          "title": "Timestamp Column",
          "type": "string"
        },
        "expire_in_days": {
          "description": "Number of days after which data expires",
          "title": "Expire In Days",
          "type": "integer"
        }
      },
      "required": [
        "timestamp_column",
        "expire_in_days"
      ],
      "title": "AutoTTLConfig",
      "type": "object"
    },
    "Column": {
      "description": "Column definition for a model.",
      "properties": {
        "name": {
          "description": "Column name",
          "title": "Name",
          "type": "string"
        },
        "description": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Human-readable description of the column",
          "title": "Description"
        },
        "data_type": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "SQL data type of the column",
          "title": "Data Type"
        },
        "nullable": {
          "default": true,
          "description": "Whether the column allows NULL values",
          "title": "Nullable",
          "type": "boolean"
        },
        "generated": {
          "anyOf": [
            {
              "discriminator": {
                "mapping": {
                  "expression": "#/$defs/GeneratedExpressionColumnConfig",
                  "identity": "#/$defs/GeneratedIdentityColumnConfig"
                },
                "propertyName": "type"
              },
              "oneOf": [
                {
                  "$ref": "#/$defs/GeneratedIdentityColumnConfig"
                },
                {
                  "$ref": "#/$defs/GeneratedExpressionColumnConfig"
                }
              ]
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Configuration for generated columns (identity or expression based)",
          "title": "Generated"
        },
        "tags": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "Metadata tags for the column",
          "title": "Tags",
          "type": "object"
        },
        "meta": {
          "additionalProperties": true,
          "description": "Generic user-defined metadata for filtering and grouping columns",
          "title": "Meta",
          "type": "object"
        }
      },
      "required": [
        "name"
      ],
      "title": "Column",
      "type": "object"
    },
    "ColumnSelector": {
      "description": "Include/exclude selector for a set of columns.\n\nExactly one of ``include`` or ``exclude`` may be set. Matching is\ncase-insensitive.\n\nArgs:\n    include: Only these columns are selected.\n    exclude: All columns except these are selected.",
      "properties": {
        "include": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Only these columns are selected (case-insensitive).",
          "title": "Include"
        },
        "exclude": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "All columns except these are selected (case-insensitive).",
          "title": "Exclude"
        }
      },
      "title": "ColumnSelector",
      "type": "object"
    },
    "DQXQuality": {
      "properties": {
        "engine": {
          "const": "dqx",
          "description": "Quality engine type",
          "title": "Engine",
          "type": "string"
        },
        "level": {
          "default": "row",
          "description": "Level at which quality is enforced",
          "enum": [
            "row",
            "table"
          ],
          "title": "Level",
          "type": "string"
        },
        "sdp_expect_level": {
          "default": "warn",
          "description": "Action for quality violations: warn, fail, drop, or deactivate",
          "enum": [
            "warn",
            "fail",
            "drop",
            "deactivate"
          ],
          "title": "Sdp Expect Level",
          "type": "string"
        },
        "sdp_quarantine": {
          "default": false,
          "description": "Whether to quarantine rows failing quality checks",
          "title": "Sdp Quarantine",
          "type": "boolean"
        },
        "spark_violation_action": {
          "default": "error",
          "description": "Action for quality violations at the Spark level: error (raise exception), ignore (store errors in output), or drop (exclude from output)",
          "enum": [
            "error",
            "ignore",
            "drop"
          ],
          "title": "Spark Violation Action",
          "type": "string"
        },
        "spark_quarantine": {
          "default": false,
          "description": "Whether to write rows failing quality checks to a quarantine table at the Spark level",
          "title": "Spark Quarantine",
          "type": "boolean"
        },
        "checks": {
          "description": "Quality check configurations",
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "title": "Checks",
          "type": "array"
        }
      },
      "required": [
        "engine"
      ],
      "title": "DQXQuality",
      "type": "object"
    },
    "ForeignKeyConstraint": {
      "properties": {
        "name": {
          "description": "Constraint name",
          "title": "Name",
          "type": "string"
        },
        "type": {
          "default": "foreign_key",
          "description": "Constraint type identifier",
          "title": "Type",
          "type": "string"
        },
        "columns": {
          "description": "List of local column names",
          "items": {
            "type": "string"
          },
          "title": "Columns",
          "type": "array"
        },
        "reference_table": {
          "description": "Fully qualified name of the referenced table",
          "title": "Reference Table",
          "type": "string"
        },
        "reference_columns": {
          "description": "List of column names in the referenced table",
          "items": {
            "type": "string"
          },
          "title": "Reference Columns",
          "type": "array"
        }
      },
      "required": [
        "name",
        "reference_table"
      ],
      "title": "ForeignKeyConstraint",
      "type": "object"
    },
    "GeneratedExpressionColumnConfig": {
      "properties": {
        "type": {
          "const": "expression",
          "description": "Column type identifier",
          "title": "Type",
          "type": "string"
        },
        "expression": {
          "description": "SQL expression used to generate the column value",
          "title": "Expression",
          "type": "string"
        }
      },
      "required": [
        "type",
        "expression"
      ],
      "title": "GeneratedExpressionColumnConfig",
      "type": "object"
    },
    "GeneratedIdentityColumnConfig": {
      "properties": {
        "type": {
          "const": "identity",
          "description": "Column type identifier",
          "title": "Type",
          "type": "string"
        },
        "as_default": {
          "default": false,
          "description": "Generated as default (True) or always (False)",
          "title": "As Default",
          "type": "boolean"
        },
        "start_with": {
          "default": 1,
          "description": "Starting value for the identity sequence",
          "title": "Start With",
          "type": "integer"
        },
        "increment_by": {
          "default": 1,
          "description": "Increment step for the identity sequence",
          "title": "Increment By",
          "type": "integer"
        }
      },
      "required": [
        "type"
      ],
      "title": "GeneratedIdentityColumnConfig",
      "type": "object"
    },
    "MergeConfig": {
      "additionalProperties": false,
      "description": "Merge rows by key, keeping one current version per key (SCD type 1).",
      "properties": {
        "options": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
          "title": "Options",
          "type": "object"
        },
        "allow_full_refresh": {
          "default": true,
          "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
          "title": "Allow Full Refresh",
          "type": "boolean"
        },
        "keys": {
          "description": "Business key columns identifying a row. Required.",
          "items": {
            "type": "string"
          },
          "minItems": 1,
          "title": "Keys",
          "type": "array"
        },
        "sequence_by": {
          "description": "Columns ordering source rows in time. Multiple columns are compared as a struct. Used to deduplicate the batch and to ignore out-of-order rows.",
          "items": {
            "type": "string"
          },
          "title": "Sequence By",
          "type": "array"
        },
        "columns": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which source columns reach the target. Keys are always included."
        },
        "track_changes": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys."
        },
        "when_deleted": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "SQL predicate marking source rows as deletes (CDC tombstones).",
          "title": "When Deleted"
        },
        "where": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Predicate narrowing the target rows taken into account.",
          "title": "Where"
        },
        "ignore_null_updates": {
          "default": false,
          "description": "Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.",
          "title": "Ignore Null Updates",
          "type": "boolean"
        },
        "ignore_null_updates_columns": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which columns ignore_null_updates applies to. Defaults to every written column except the keys."
        },
        "schema_evolution": {
          "default": true,
          "description": "Whether new source columns are added to the target during the merge.",
          "title": "Schema Evolution",
          "type": "boolean"
        },
        "mode": {
          "const": "merge",
          "default": "merge",
          "title": "Mode",
          "type": "string"
        },
        "sql_conditions": {
          "anyOf": [
            {
              "$ref": "#/$defs/SqlConditions"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Raw SQL conditions for the merge clauses."
        },
        "insert_only_columns": {
          "description": "Columns written on insert but left out of the update, so their first value survives (e.g. created_at). Unlike track_changes, which decides whether an update happens at all, this decides which columns an update may touch.",
          "items": {
            "type": "string"
          },
          "title": "Insert Only Columns",
          "type": "array"
        },
        "missing_in_source": {
          "default": "ignore",
          "description": "What to do with target rows that the source does not contain.",
          "enum": [
            "ignore",
            "delete"
          ],
          "title": "Missing In Source",
          "type": "string"
        }
      },
      "required": [
        "keys"
      ],
      "title": "MergeConfig",
      "type": "object"
    },
    "OverwriteConfig": {
      "additionalProperties": false,
      "description": "Replace the target table contents with the DataFrame.\n\nArgs:\n    replace_where: Optional predicate limiting the overwrite to matching rows\n        (Delta ``replaceWhere``) instead of the whole table.",
      "properties": {
        "options": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
          "title": "Options",
          "type": "object"
        },
        "allow_full_refresh": {
          "default": true,
          "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
          "title": "Allow Full Refresh",
          "type": "boolean"
        },
        "mode": {
          "const": "overwrite",
          "default": "overwrite",
          "title": "Mode",
          "type": "string"
        },
        "replace_where": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Predicate limiting the overwrite to matching rows (Delta replaceWhere).",
          "title": "Replace Where"
        }
      },
      "title": "OverwriteConfig",
      "type": "object"
    },
    "PrimaryKeyConstraint": {
      "properties": {
        "name": {
          "description": "Constraint name",
          "title": "Name",
          "type": "string"
        },
        "type": {
          "default": "primary_key",
          "description": "Constraint type identifier",
          "title": "Type",
          "type": "string"
        },
        "columns": {
          "description": "List of column names forming the primary key",
          "items": {
            "type": "string"
          },
          "title": "Columns",
          "type": "array"
        }
      },
      "required": [
        "name"
      ],
      "title": "PrimaryKeyConstraint",
      "type": "object"
    },
    "SDPQuality": {
      "properties": {
        "engine": {
          "const": "sdp",
          "description": "Quality engine type",
          "title": "Engine",
          "type": "string"
        },
        "level": {
          "const": "row",
          "default": "row",
          "description": "Quality enforcement level",
          "title": "Level",
          "type": "string"
        },
        "expect_all": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "SQL expressions that must pass",
          "title": "Expect All",
          "type": "object"
        },
        "expect_all_or_drop": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "SQL expressions; failing rows are dropped",
          "title": "Expect All Or Drop",
          "type": "object"
        },
        "expect_all_or_fail": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "SQL expressions; job fails if any expression fails",
          "title": "Expect All Or Fail",
          "type": "object"
        },
        "expect_all_or_quarantine": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "SQL expressions; failing rows are quarantined",
          "title": "Expect All Or Quarantine",
          "type": "object"
        }
      },
      "required": [
        "engine"
      ],
      "title": "SDPQuality",
      "type": "object"
    },
    "Scd2Columns": {
      "description": "Names of the history-tracking columns maintained by ``mode: scd2``.\n\nDefaults match Databricks AUTO CDC (``apply_changes``) so SCD2 tables stay\ninterchangeable between SDP pipelines and kelp Spark jobs.\n\nArgs:\n    valid_from: Column holding the sequence value a version becomes valid at.\n    valid_to: Column holding the sequence value a version is superseded at.\n    is_current: Optional boolean column maintained alongside ``valid_to``.\n    open_value: SQL expression ``valid_to`` takes while a version is current,\n        instead of ``NULL`` \u2014 e.g. ``\"'2999-12-31'\"`` or ``\"9999999999\"``.",
      "properties": {
        "valid_from": {
          "default": "__START_AT",
          "description": "Column holding the sequence value a version becomes valid at.",
          "title": "Valid From",
          "type": "string"
        },
        "valid_to": {
          "default": "__END_AT",
          "description": "Column holding the sequence value a version is superseded at (NULL for the current version unless open_value is set).",
          "title": "Valid To",
          "type": "string"
        },
        "is_current": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Optional boolean column maintained alongside valid_to.",
          "title": "Is Current"
        },
        "open_value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "SQL expression valid_to takes while a version is current, instead of NULL (e.g. \"'2999-12-31'\"). Must be castable to the sequence_by type.",
          "title": "Open Value"
        }
      },
      "title": "Scd2Columns",
      "type": "object"
    },
    "Scd2Config": {
      "additionalProperties": false,
      "description": "Track full row history by key, closing superseded versions (SCD type 2).",
      "properties": {
        "options": {
          "additionalProperties": {
            "type": "string"
          },
          "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
          "title": "Options",
          "type": "object"
        },
        "allow_full_refresh": {
          "default": true,
          "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
          "title": "Allow Full Refresh",
          "type": "boolean"
        },
        "keys": {
          "description": "Business key columns identifying a row. Required.",
          "items": {
            "type": "string"
          },
          "minItems": 1,
          "title": "Keys",
          "type": "array"
        },
        "sequence_by": {
          "description": "Columns ordering source rows in time. Required for scd2: they become the valid_from/valid_to interval bounds.",
          "items": {
            "type": "string"
          },
          "minItems": 1,
          "title": "Sequence By",
          "type": "array"
        },
        "columns": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which source columns reach the target. Keys are always included."
        },
        "track_changes": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys."
        },
        "when_deleted": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "SQL predicate marking source rows as deletes (CDC tombstones).",
          "title": "When Deleted"
        },
        "where": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Predicate narrowing the target rows taken into account.",
          "title": "Where"
        },
        "ignore_null_updates": {
          "default": false,
          "description": "Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.",
          "title": "Ignore Null Updates",
          "type": "boolean"
        },
        "ignore_null_updates_columns": {
          "anyOf": [
            {
              "$ref": "#/$defs/ColumnSelector"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Which columns ignore_null_updates applies to. Defaults to every written column except the keys."
        },
        "schema_evolution": {
          "default": true,
          "description": "Whether new source columns are added to the target during the merge.",
          "title": "Schema Evolution",
          "type": "boolean"
        },
        "mode": {
          "const": "scd2",
          "default": "scd2",
          "title": "Mode",
          "type": "string"
        },
        "history": {
          "$ref": "#/$defs/Scd2Columns",
          "description": "Names of the history-tracking columns kelp maintains."
        }
      },
      "required": [
        "keys",
        "sequence_by"
      ],
      "title": "Scd2Config",
      "type": "object"
    },
    "SqlConditions": {
      "description": "Raw SQL conditions for the merge clauses.\n\nEscape hatch for cases the declarative fields cannot express. Every entry is a\nboolean SQL expression guarding one merge clause and may reference the\n``source`` and ``target`` aliases \u2014 nothing else about the merge is settable here.\n\nArgs:\n    when_matched: Replaces the derived \"row changed\" condition on updates.\n    when_not_matched: Extra condition applied to inserts of unmatched source rows.\n    when_not_matched_by_source: Extra condition applied to target rows missing\n        from the source. Requires ``missing_in_source: delete``.",
      "properties": {
        "when_matched": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Replaces the derived 'row changed' condition on matched updates.",
          "title": "When Matched"
        },
        "when_not_matched": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Extra condition applied when inserting unmatched source rows.",
          "title": "When Not Matched"
        },
        "when_not_matched_by_source": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Extra condition applied to target rows missing from the source. Requires missing_in_source='delete'.",
          "title": "When Not Matched By Source"
        }
      },
      "title": "SqlConditions",
      "type": "object"
    },
    "TableType": {
      "enum": [
        "external",
        "external_shallow_clone",
        "foreign",
        "managed",
        "managed_shallow_clone",
        "materialized_view",
        "metric_view",
        "streaming_table",
        "view"
      ],
      "title": "TableType",
      "type": "string"
    }
  },
  "description": "Model definition in Unity Catalog.\n\nRepresents a metadata model in Kelp with configuration that maps to\nDatabricks Unity Catalog objects.",
  "properties": {
    "table_type": {
      "$ref": "#/$defs/TableType",
      "default": "managed",
      "description": "Type of table: managed, external, view, streaming_table, etc."
    },
    "catalog": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Unity Catalog name",
      "title": "Catalog"
    },
    "schema": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Schema/database name",
      "title": "Schema"
    },
    "name": {
      "description": "Model name",
      "title": "Name",
      "type": "string"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Human-readable description of the model",
      "title": "Description"
    },
    "spark_conf": {
      "additionalProperties": true,
      "description": "Spark configuration properties",
      "title": "Spark Conf",
      "type": "object"
    },
    "table_properties": {
      "additionalProperties": true,
      "description": "Databricks table properties",
      "title": "Table Properties",
      "type": "object"
    },
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Physical path for external tables or custom locations",
      "title": "Path"
    },
    "partition_cols": {
      "description": "List of column names for partitioning",
      "items": {
        "type": "string"
      },
      "title": "Partition Cols",
      "type": "array"
    },
    "cluster_by_auto": {
      "default": false,
      "description": "Enable automatic clustering optimization",
      "title": "Cluster By Auto",
      "type": "boolean"
    },
    "cluster_by": {
      "description": "List of column names for explicit clustering (max 4)",
      "items": {
        "type": "string"
      },
      "maxItems": 4,
      "title": "Cluster By",
      "type": "array"
    },
    "row_filter": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "SQL expression to filter rows based on security policies",
      "title": "Row Filter"
    },
    "columns": {
      "description": "Column definitions for the model",
      "items": {
        "$ref": "#/$defs/Column"
      },
      "title": "Columns",
      "type": "array"
    },
    "quality": {
      "anyOf": [
        {
          "discriminator": {
            "mapping": {
              "dqx": "#/$defs/DQXQuality",
              "sdp": "#/$defs/SDPQuality"
            },
            "propertyName": "engine"
          },
          "oneOf": [
            {
              "$ref": "#/$defs/SDPQuality"
            },
            {
              "$ref": "#/$defs/DQXQuality"
            }
          ]
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Data quality configuration using SDPQuality or DQXQuality",
      "title": "Quality"
    },
    "materialization": {
      "anyOf": [
        {
          "discriminator": {
            "mapping": {
              "append": "#/$defs/AppendConfig",
              "merge": "#/$defs/MergeConfig",
              "overwrite": "#/$defs/OverwriteConfig",
              "scd2": "#/$defs/Scd2Config"
            },
            "propertyName": "mode"
          },
          "oneOf": [
            {
              "$ref": "#/$defs/AppendConfig"
            },
            {
              "$ref": "#/$defs/OverwriteConfig"
            },
            {
              "$ref": "#/$defs/MergeConfig"
            },
            {
              "$ref": "#/$defs/Scd2Config"
            }
          ]
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Configuration controlling how the model is materialized.",
      "title": "Materialization"
    },
    "constraints": {
      "description": "Constraints like primary key or foreign key",
      "items": {
        "anyOf": [
          {
            "$ref": "#/$defs/PrimaryKeyConstraint"
          },
          {
            "$ref": "#/$defs/ForeignKeyConstraint"
          }
        ]
      },
      "title": "Constraints",
      "type": "array"
    },
    "auto_ttl": {
      "anyOf": [
        {
          "$ref": "#/$defs/AutoTTLConfig"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Configuration for automatic TTL (Time To Live) based on a timestamp column"
    },
    "tags": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Metadata tags for the model",
      "title": "Tags",
      "type": "object"
    },
    "meta": {
      "additionalProperties": true,
      "description": "Generic user-defined metadata for filtering and grouping",
      "title": "Meta",
      "type": "object"
    }
  },
  "required": [
    "name"
  ],
  "title": "Model",
  "type": "object"
}

Config:

• validate_by_name: True
• validate_by_alias: True
• serialize_by_alias: True
• use_enum_values: True

Fields:

• origin_file_path (SkipJsonSchema[str] | None)
• table_type (TableType)
• catalog (str | None)
• schema_ (str | None)
• name (str)
• description (str | None)
• spark_conf (dict)
• table_properties (dict)
• path (str | None)
• partition_cols (list[str])
• cluster_by_auto (bool)
• cluster_by (list[str])
• row_filter (str | None)
• columns (list[Column])
• quality (SDPQuality | DQXQuality | None)
• materialization (MaterializationConfig | None)
• constraints (list[PrimaryKeyConstraint | ForeignKeyConstraint])
• auto_ttl (AutoTTLConfig | None)
• tags (dict[str, str])
• meta (dict[str, Any])
• raw_config (SkipJsonSchema[dict])

Validators:

• _default_materialization_mode → materialization
• _serialize_complex_property_values → table_properties
• _validate_catalog_requires_schema

origin_file_path pydantic-field

origin_file_path = None

Path to the source YAML file defining this model

table_type pydantic-field

table_type = TableType.MANAGED

Type of table: managed, external, view, streaming_table, etc.

catalog pydantic-field

catalog = None

Unity Catalog name

schema_ pydantic-field

schema_ = None

Schema/database name

name pydantic-field

name

Model name

description pydantic-field

description = None

Human-readable description of the model

spark_conf pydantic-field

spark_conf

Spark configuration properties

table_properties pydantic-field

table_properties

Databricks table properties

path pydantic-field

path = None

Physical path for external tables or custom locations

partition_cols pydantic-field

partition_cols

List of column names for partitioning

cluster_by_auto pydantic-field

cluster_by_auto = False

Enable automatic clustering optimization

cluster_by pydantic-field

cluster_by

List of column names for explicit clustering (max 4)

row_filter pydantic-field

row_filter = None

SQL expression to filter rows based on security policies

columns pydantic-field

columns

Column definitions for the model

quality pydantic-field

quality = None

Data quality configuration using SDPQuality or DQXQuality

materialization pydantic-field

materialization = None

Configuration controlling how the model is materialized.

constraints pydantic-field

constraints

Constraints like primary key or foreign key

auto_ttl pydantic-field

auto_ttl = None

Configuration for automatic TTL (Time To Live) based on a timestamp column

tags pydantic-field

tags

Metadata tags for the model

meta pydantic-field

meta

Generic user-defined metadata for filtering and grouping

raw_config pydantic-field

raw_config

Original unparsed configuration preserving placeholder variables

model_config class-attribute instance-attribute

model_config = ConfigDict(
    validate_by_name=True,
    validate_by_alias=True,
    serialize_by_alias=True,
    use_enum_values=True,
)

deserialize_property_values staticmethod

deserialize_property_values(properties)

Deserialize JSON-encoded property values back to complex types.

Used by the YAML writer to restore the original structure (list, dict) so that the YAML output uses native YAML types instead of JSON strings.

Source code in src/kelp/models/model.py

@staticmethod
def deserialize_property_values(properties: dict) -> dict:
    """Deserialize JSON-encoded property values back to complex types.

    Used by the YAML writer to restore the original structure (list, dict)
    so that the YAML output uses native YAML types instead of JSON strings.
    """
    if not isinstance(properties, dict):
        return properties
    result: dict = {}
    for key, value in properties.items():
        if isinstance(value, str):
            try:
                parsed = json.loads(value)
            except (json.JSONDecodeError, ValueError):
                parsed = value
            if isinstance(parsed, (dict, list)):
                result[key] = parsed
            else:
                result[key] = value
        else:
            result[key] = value
    return result

get_qualified_name

get_qualified_name()

Get the fully qualified model name including database/schema if applicable.

Source code in src/kelp/models/model.py

def get_qualified_name(self) -> str:
    """Get the fully qualified model name including database/schema if applicable."""
    parts = []
    if self.catalog:
        parts.append(self.catalog)
    if self.schema_:
        parts.append(self.schema_)
    parts.append(self.name)
    return ".".join(parts)

kelp.models.model_mat_config.AppendConfig pydantic-model

Bases: _BaseMaterialization

Append the DataFrame to the target table.

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Append the DataFrame to the target table.",
  "properties": {
    "options": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
      "title": "Options",
      "type": "object"
    },
    "allow_full_refresh": {
      "default": true,
      "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
      "title": "Allow Full Refresh",
      "type": "boolean"
    },
    "mode": {
      "const": "append",
      "default": "append",
      "title": "Mode",
      "type": "string"
    }
  },
  "title": "AppendConfig",
  "type": "object"
}

Fields:

• options (dict[str, str])
• allow_full_refresh (bool)
• mode (Literal['append'])

mode pydantic-field

mode = 'append'

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

options pydantic-field

options

Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).

allow_full_refresh pydantic-field

allow_full_refresh = True

Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.

kelp.models.model_mat_config.OverwriteConfig pydantic-model

Bases: _BaseMaterialization

Replace the target table contents with the DataFrame.

Parameters:

Name	Type	Description	Default
replace_where		Optional predicate limiting the overwrite to matching rows (Delta replaceWhere) instead of the whole table.	required

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Replace the target table contents with the DataFrame.\n\nArgs:\n    replace_where: Optional predicate limiting the overwrite to matching rows\n        (Delta ``replaceWhere``) instead of the whole table.",
  "properties": {
    "options": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
      "title": "Options",
      "type": "object"
    },
    "allow_full_refresh": {
      "default": true,
      "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
      "title": "Allow Full Refresh",
      "type": "boolean"
    },
    "mode": {
      "const": "overwrite",
      "default": "overwrite",
      "title": "Mode",
      "type": "string"
    },
    "replace_where": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Predicate limiting the overwrite to matching rows (Delta replaceWhere).",
      "title": "Replace Where"
    }
  },
  "title": "OverwriteConfig",
  "type": "object"
}

Fields:

• options (dict[str, str])
• allow_full_refresh (bool)
• mode (Literal['overwrite'])
• replace_where (str | None)

mode pydantic-field

mode = 'overwrite'

replace_where pydantic-field

replace_where = None

Predicate limiting the overwrite to matching rows (Delta replaceWhere).

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

options pydantic-field

options

Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).

allow_full_refresh pydantic-field

allow_full_refresh = True

Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.

kelp.models.model_mat_config.MergeConfig pydantic-model

Bases: _BaseMerge

Merge rows by key, keeping one current version per key (SCD type 1).

Show JSON schema:

{
  "$defs": {
    "ColumnSelector": {
      "description": "Include/exclude selector for a set of columns.\n\nExactly one of ``include`` or ``exclude`` may be set. Matching is\ncase-insensitive.\n\nArgs:\n    include: Only these columns are selected.\n    exclude: All columns except these are selected.",
      "properties": {
        "include": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Only these columns are selected (case-insensitive).",
          "title": "Include"
        },
        "exclude": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "All columns except these are selected (case-insensitive).",
          "title": "Exclude"
        }
      },
      "title": "ColumnSelector",
      "type": "object"
    },
    "SqlConditions": {
      "description": "Raw SQL conditions for the merge clauses.\n\nEscape hatch for cases the declarative fields cannot express. Every entry is a\nboolean SQL expression guarding one merge clause and may reference the\n``source`` and ``target`` aliases \u2014 nothing else about the merge is settable here.\n\nArgs:\n    when_matched: Replaces the derived \"row changed\" condition on updates.\n    when_not_matched: Extra condition applied to inserts of unmatched source rows.\n    when_not_matched_by_source: Extra condition applied to target rows missing\n        from the source. Requires ``missing_in_source: delete``.",
      "properties": {
        "when_matched": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Replaces the derived 'row changed' condition on matched updates.",
          "title": "When Matched"
        },
        "when_not_matched": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Extra condition applied when inserting unmatched source rows.",
          "title": "When Not Matched"
        },
        "when_not_matched_by_source": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Extra condition applied to target rows missing from the source. Requires missing_in_source='delete'.",
          "title": "When Not Matched By Source"
        }
      },
      "title": "SqlConditions",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Merge rows by key, keeping one current version per key (SCD type 1).",
  "properties": {
    "options": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
      "title": "Options",
      "type": "object"
    },
    "allow_full_refresh": {
      "default": true,
      "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
      "title": "Allow Full Refresh",
      "type": "boolean"
    },
    "keys": {
      "description": "Business key columns identifying a row. Required.",
      "items": {
        "type": "string"
      },
      "minItems": 1,
      "title": "Keys",
      "type": "array"
    },
    "sequence_by": {
      "description": "Columns ordering source rows in time. Multiple columns are compared as a struct. Used to deduplicate the batch and to ignore out-of-order rows.",
      "items": {
        "type": "string"
      },
      "title": "Sequence By",
      "type": "array"
    },
    "columns": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which source columns reach the target. Keys are always included."
    },
    "track_changes": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys."
    },
    "when_deleted": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "SQL predicate marking source rows as deletes (CDC tombstones).",
      "title": "When Deleted"
    },
    "where": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Predicate narrowing the target rows taken into account.",
      "title": "Where"
    },
    "ignore_null_updates": {
      "default": false,
      "description": "Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.",
      "title": "Ignore Null Updates",
      "type": "boolean"
    },
    "ignore_null_updates_columns": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which columns ignore_null_updates applies to. Defaults to every written column except the keys."
    },
    "schema_evolution": {
      "default": true,
      "description": "Whether new source columns are added to the target during the merge.",
      "title": "Schema Evolution",
      "type": "boolean"
    },
    "mode": {
      "const": "merge",
      "default": "merge",
      "title": "Mode",
      "type": "string"
    },
    "sql_conditions": {
      "anyOf": [
        {
          "$ref": "#/$defs/SqlConditions"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Raw SQL conditions for the merge clauses."
    },
    "insert_only_columns": {
      "description": "Columns written on insert but left out of the update, so their first value survives (e.g. created_at). Unlike track_changes, which decides whether an update happens at all, this decides which columns an update may touch.",
      "items": {
        "type": "string"
      },
      "title": "Insert Only Columns",
      "type": "array"
    },
    "missing_in_source": {
      "default": "ignore",
      "description": "What to do with target rows that the source does not contain.",
      "enum": [
        "ignore",
        "delete"
      ],
      "title": "Missing In Source",
      "type": "string"
    }
  },
  "required": [
    "keys"
  ],
  "title": "MergeConfig",
  "type": "object"
}

Fields:

• options (dict[str, str])
• allow_full_refresh (bool)
• keys (list[str])
• sequence_by (list[str])
• columns (ColumnSelector | None)
• track_changes (ColumnSelector | None)
• when_deleted (str | None)
• where (str | None)
• ignore_null_updates (bool)
• ignore_null_updates_columns (ColumnSelector | None)
• schema_evolution (bool)
• mode (Literal['merge'])
• sql_conditions (SqlConditions | None)
• insert_only_columns (list[str])
• missing_in_source (Literal['ignore', 'delete'])

mode pydantic-field

mode = 'merge'

sql_conditions pydantic-field

sql_conditions = None

Raw SQL conditions for the merge clauses.

insert_only_columns pydantic-field

insert_only_columns

Columns written on insert but left out of the update, so their first value survives (e.g. created_at). Unlike track_changes, which decides whether an update happens at all, this decides which columns an update may touch.

missing_in_source pydantic-field

missing_in_source = 'ignore'

What to do with target rows that the source does not contain.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

options pydantic-field

options

Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).

allow_full_refresh pydantic-field

allow_full_refresh = True

Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.

keys pydantic-field

keys

Business key columns identifying a row. Required.

sequence_by pydantic-field

sequence_by

Columns ordering source rows in time. Multiple columns are compared as a struct. Used to deduplicate the batch and to ignore out-of-order rows.

columns pydantic-field

columns = None

Which source columns reach the target. Keys are always included.

track_changes pydantic-field

track_changes = None

Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys.

when_deleted pydantic-field

when_deleted = None

SQL predicate marking source rows as deletes (CDC tombstones).

where pydantic-field

where = None

Predicate narrowing the target rows taken into account.

ignore_null_updates pydantic-field

ignore_null_updates = False

Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.

ignore_null_updates_columns pydantic-field

ignore_null_updates_columns = None

Which columns ignore_null_updates applies to. Defaults to every written column except the keys.

schema_evolution pydantic-field

schema_evolution = True

Whether new source columns are added to the target during the merge.

ignore_null_columns

ignore_null_columns(candidates)

Return the columns whose NULL source values must not replace stored values.

Parameters:

Name	Type	Description	Default
candidates	list[str]	Columns eligible for the rule, normally the written columns except the keys.	required

Returns:

Type	Description
list[str]	Selected column names, empty when ignore_null_updates is off.

Source code in src/kelp/models/model_mat_config.py

def ignore_null_columns(self, candidates: list[str]) -> list[str]:
    """Return the columns whose NULL source values must not replace stored values.

    Args:
        candidates: Columns eligible for the rule, normally the written columns
            except the keys.

    Returns:
        Selected column names, empty when ``ignore_null_updates`` is off.
    """
    if not self.ignore_null_updates:
        return []
    return (self.ignore_null_updates_columns or ColumnSelector()).apply(candidates)

kelp.models.model_mat_config.Scd2Config pydantic-model

Bases: _BaseMerge

Track full row history by key, closing superseded versions (SCD type 2).

Show JSON schema:

{
  "$defs": {
    "ColumnSelector": {
      "description": "Include/exclude selector for a set of columns.\n\nExactly one of ``include`` or ``exclude`` may be set. Matching is\ncase-insensitive.\n\nArgs:\n    include: Only these columns are selected.\n    exclude: All columns except these are selected.",
      "properties": {
        "include": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Only these columns are selected (case-insensitive).",
          "title": "Include"
        },
        "exclude": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "All columns except these are selected (case-insensitive).",
          "title": "Exclude"
        }
      },
      "title": "ColumnSelector",
      "type": "object"
    },
    "Scd2Columns": {
      "description": "Names of the history-tracking columns maintained by ``mode: scd2``.\n\nDefaults match Databricks AUTO CDC (``apply_changes``) so SCD2 tables stay\ninterchangeable between SDP pipelines and kelp Spark jobs.\n\nArgs:\n    valid_from: Column holding the sequence value a version becomes valid at.\n    valid_to: Column holding the sequence value a version is superseded at.\n    is_current: Optional boolean column maintained alongside ``valid_to``.\n    open_value: SQL expression ``valid_to`` takes while a version is current,\n        instead of ``NULL`` \u2014 e.g. ``\"'2999-12-31'\"`` or ``\"9999999999\"``.",
      "properties": {
        "valid_from": {
          "default": "__START_AT",
          "description": "Column holding the sequence value a version becomes valid at.",
          "title": "Valid From",
          "type": "string"
        },
        "valid_to": {
          "default": "__END_AT",
          "description": "Column holding the sequence value a version is superseded at (NULL for the current version unless open_value is set).",
          "title": "Valid To",
          "type": "string"
        },
        "is_current": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Optional boolean column maintained alongside valid_to.",
          "title": "Is Current"
        },
        "open_value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "SQL expression valid_to takes while a version is current, instead of NULL (e.g. \"'2999-12-31'\"). Must be castable to the sequence_by type.",
          "title": "Open Value"
        }
      },
      "title": "Scd2Columns",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Track full row history by key, closing superseded versions (SCD type 2).",
  "properties": {
    "options": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).",
      "title": "Options",
      "type": "object"
    },
    "allow_full_refresh": {
      "default": true,
      "description": "Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.",
      "title": "Allow Full Refresh",
      "type": "boolean"
    },
    "keys": {
      "description": "Business key columns identifying a row. Required.",
      "items": {
        "type": "string"
      },
      "minItems": 1,
      "title": "Keys",
      "type": "array"
    },
    "sequence_by": {
      "description": "Columns ordering source rows in time. Required for scd2: they become the valid_from/valid_to interval bounds.",
      "items": {
        "type": "string"
      },
      "minItems": 1,
      "title": "Sequence By",
      "type": "array"
    },
    "columns": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which source columns reach the target. Keys are always included."
    },
    "track_changes": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys."
    },
    "when_deleted": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "SQL predicate marking source rows as deletes (CDC tombstones).",
      "title": "When Deleted"
    },
    "where": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Predicate narrowing the target rows taken into account.",
      "title": "Where"
    },
    "ignore_null_updates": {
      "default": false,
      "description": "Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.",
      "title": "Ignore Null Updates",
      "type": "boolean"
    },
    "ignore_null_updates_columns": {
      "anyOf": [
        {
          "$ref": "#/$defs/ColumnSelector"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Which columns ignore_null_updates applies to. Defaults to every written column except the keys."
    },
    "schema_evolution": {
      "default": true,
      "description": "Whether new source columns are added to the target during the merge.",
      "title": "Schema Evolution",
      "type": "boolean"
    },
    "mode": {
      "const": "scd2",
      "default": "scd2",
      "title": "Mode",
      "type": "string"
    },
    "history": {
      "$ref": "#/$defs/Scd2Columns",
      "description": "Names of the history-tracking columns kelp maintains."
    }
  },
  "required": [
    "keys",
    "sequence_by"
  ],
  "title": "Scd2Config",
  "type": "object"
}

Fields:

• options (dict[str, str])
• allow_full_refresh (bool)
• keys (list[str])
• columns (ColumnSelector | None)
• track_changes (ColumnSelector | None)
• when_deleted (str | None)
• where (str | None)
• ignore_null_updates (bool)
• ignore_null_updates_columns (ColumnSelector | None)
• schema_evolution (bool)
• mode (Literal['scd2'])
• sequence_by (list[str])
• history (Scd2Columns)

mode pydantic-field

mode = 'scd2'

sequence_by pydantic-field

sequence_by

Columns ordering source rows in time. Required for scd2: they become the valid_from/valid_to interval bounds.

history pydantic-field

history

Names of the history-tracking columns kelp maintains.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

options pydantic-field

options

Extra Delta writer/merge options (e.g. {'mergeSchema': 'true'}).

allow_full_refresh pydantic-field

allow_full_refresh = True

Whether a caller-requested full refresh may drop and rebuild the target. Set to False to protect critical tables; the refresh is then skipped with a warning.

keys pydantic-field

keys

Business key columns identifying a row. Required.

columns pydantic-field

columns = None

Which source columns reach the target. Keys are always included.

track_changes pydantic-field

track_changes = None

Which columns are compared to decide whether a row changed at all. When none of them differ, nothing is written: no update (merge) and no new version (scd2). Defaults to every written column except the keys.

when_deleted pydantic-field

when_deleted = None

SQL predicate marking source rows as deletes (CDC tombstones).

where pydantic-field

where = None

Predicate narrowing the target rows taken into account.

ignore_null_updates pydantic-field

ignore_null_updates = False

Whether a NULL source value leaves the stored value alone instead of replacing it, so partial CDC rows keep the previous value.

ignore_null_updates_columns pydantic-field

ignore_null_updates_columns = None

Which columns ignore_null_updates applies to. Defaults to every written column except the keys.

schema_evolution pydantic-field

schema_evolution = True

Whether new source columns are added to the target during the merge.

ignore_null_columns

ignore_null_columns(candidates)

Return the columns whose NULL source values must not replace stored values.

Parameters:

Name	Type	Description	Default
candidates	list[str]	Columns eligible for the rule, normally the written columns except the keys.	required

Returns:

Type	Description
list[str]	Selected column names, empty when ignore_null_updates is off.

Source code in src/kelp/models/model_mat_config.py

def ignore_null_columns(self, candidates: list[str]) -> list[str]:
    """Return the columns whose NULL source values must not replace stored values.

    Args:
        candidates: Columns eligible for the rule, normally the written columns
            except the keys.

    Returns:
        Selected column names, empty when ``ignore_null_updates`` is off.
    """
    if not self.ignore_null_updates:
        return []
    return (self.ignore_null_updates_columns or ColumnSelector()).apply(candidates)

kelp.models.model_mat_config.Scd2Columns pydantic-model

Bases: BaseModel

Names of the history-tracking columns maintained by mode: scd2.

Defaults match Databricks AUTO CDC (apply_changes) so SCD2 tables stay interchangeable between SDP pipelines and kelp Spark jobs.

Parameters:

Name	Type	Description	Default
valid_from		Column holding the sequence value a version becomes valid at.	required
valid_to		Column holding the sequence value a version is superseded at.	required
is_current		Optional boolean column maintained alongside valid_to.	required
open_value		SQL expression valid_to takes while a version is current, instead of NULL — e.g. "'2999-12-31'" or "9999999999".	required

Show JSON schema:

{
  "description": "Names of the history-tracking columns maintained by ``mode: scd2``.\n\nDefaults match Databricks AUTO CDC (``apply_changes``) so SCD2 tables stay\ninterchangeable between SDP pipelines and kelp Spark jobs.\n\nArgs:\n    valid_from: Column holding the sequence value a version becomes valid at.\n    valid_to: Column holding the sequence value a version is superseded at.\n    is_current: Optional boolean column maintained alongside ``valid_to``.\n    open_value: SQL expression ``valid_to`` takes while a version is current,\n        instead of ``NULL`` \u2014 e.g. ``\"'2999-12-31'\"`` or ``\"9999999999\"``.",
  "properties": {
    "valid_from": {
      "default": "__START_AT",
      "description": "Column holding the sequence value a version becomes valid at.",
      "title": "Valid From",
      "type": "string"
    },
    "valid_to": {
      "default": "__END_AT",
      "description": "Column holding the sequence value a version is superseded at (NULL for the current version unless open_value is set).",
      "title": "Valid To",
      "type": "string"
    },
    "is_current": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional boolean column maintained alongside valid_to.",
      "title": "Is Current"
    },
    "open_value": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "SQL expression valid_to takes while a version is current, instead of NULL (e.g. \"'2999-12-31'\"). Must be castable to the sequence_by type.",
      "title": "Open Value"
    }
  },
  "title": "Scd2Columns",
  "type": "object"
}

Fields:

• valid_from (str)
• valid_to (str)
• is_current (str | None)
• open_value (str | None)

valid_from pydantic-field

valid_from = '__START_AT'

Column holding the sequence value a version becomes valid at.

valid_to pydantic-field

valid_to = '__END_AT'

Column holding the sequence value a version is superseded at (NULL for the current version unless open_value is set).

is_current pydantic-field

is_current = None

Optional boolean column maintained alongside valid_to.

open_value pydantic-field

open_value = None

SQL expression valid_to takes while a version is current, instead of NULL (e.g. "'2999-12-31'"). Must be castable to the sequence_by type.

all_names

all_names()

Return every history column name that is configured.

Source code in src/kelp/models/model_mat_config.py

def all_names(self) -> list[str]:
    """Return every history column name that is configured."""
    names = [self.valid_from, self.valid_to]
    if self.is_current:
        names.append(self.is_current)
    return names

kelp.models.model_mat_config.ColumnSelector pydantic-model

Bases: BaseModel

Include/exclude selector for a set of columns.

Exactly one of include or exclude may be set. Matching is case-insensitive.

Parameters:

Name	Type	Description	Default
include		Only these columns are selected.	required
exclude		All columns except these are selected.	required

Show JSON schema:

{
  "description": "Include/exclude selector for a set of columns.\n\nExactly one of ``include`` or ``exclude`` may be set. Matching is\ncase-insensitive.\n\nArgs:\n    include: Only these columns are selected.\n    exclude: All columns except these are selected.",
  "properties": {
    "include": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Only these columns are selected (case-insensitive).",
      "title": "Include"
    },
    "exclude": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "All columns except these are selected (case-insensitive).",
      "title": "Exclude"
    }
  },
  "title": "ColumnSelector",
  "type": "object"
}

Fields:

• include (list[str] | None)
• exclude (list[str] | None)

Validators:

• _validate_exclusive

include pydantic-field

include = None

Only these columns are selected (case-insensitive).

exclude pydantic-field

exclude = None

All columns except these are selected (case-insensitive).

apply

apply(candidates, required=None)

Select from candidates, preserving their order.

Parameters:

Name	Type	Description	Default
candidates	list[str]	Columns to select from.	required
required	list[str] | None	Columns always kept, even when not selected.	None

Returns:

Type	Description
list[str]	Selected column names in candidates order.

Source code in src/kelp/models/model_mat_config.py

def apply(self, candidates: list[str], required: list[str] | None = None) -> list[str]:
    """Select from ``candidates``, preserving their order.

    Args:
        candidates: Columns to select from.
        required: Columns always kept, even when not selected.

    Returns:
        Selected column names in ``candidates`` order.
    """
    selected = list(candidates)
    if self.include is not None:
        wanted = {name.lower() for name in self.include}
        selected = [col for col in candidates if col.lower() in wanted]
    elif self.exclude is not None:
        unwanted = {name.lower() for name in self.exclude}
        selected = [col for col in candidates if col.lower() not in unwanted]

    selected_lower = {col.lower() for col in selected}
    lookup = {col.lower(): col for col in candidates}
    for name in required or []:
        if name.lower() in lookup and name.lower() not in selected_lower:
            selected.append(lookup[name.lower()])
            selected_lower.add(name.lower())
    return selected

names

names()

Return the configured column names, whichever side is set.

Source code in src/kelp/models/model_mat_config.py

def names(self) -> list[str]:
    """Return the configured column names, whichever side is set."""
    return list(self.include or self.exclude or [])

kelp.models.model_mat_config.SqlConditions pydantic-model

Bases: BaseModel

Raw SQL conditions for the merge clauses.

Escape hatch for cases the declarative fields cannot express. Every entry is a boolean SQL expression guarding one merge clause and may reference the source and target aliases — nothing else about the merge is settable here.

Parameters:

Name	Type	Description	Default
when_matched		Replaces the derived "row changed" condition on updates.	required
when_not_matched		Extra condition applied to inserts of unmatched source rows.	required
when_not_matched_by_source		Extra condition applied to target rows missing from the source. Requires missing_in_source: delete.	required

Show JSON schema:

{
  "description": "Raw SQL conditions for the merge clauses.\n\nEscape hatch for cases the declarative fields cannot express. Every entry is a\nboolean SQL expression guarding one merge clause and may reference the\n``source`` and ``target`` aliases \u2014 nothing else about the merge is settable here.\n\nArgs:\n    when_matched: Replaces the derived \"row changed\" condition on updates.\n    when_not_matched: Extra condition applied to inserts of unmatched source rows.\n    when_not_matched_by_source: Extra condition applied to target rows missing\n        from the source. Requires ``missing_in_source: delete``.",
  "properties": {
    "when_matched": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Replaces the derived 'row changed' condition on matched updates.",
      "title": "When Matched"
    },
    "when_not_matched": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Extra condition applied when inserting unmatched source rows.",
      "title": "When Not Matched"
    },
    "when_not_matched_by_source": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Extra condition applied to target rows missing from the source. Requires missing_in_source='delete'.",
      "title": "When Not Matched By Source"
    }
  },
  "title": "SqlConditions",
  "type": "object"
}

Fields:

• when_matched (str | None)
• when_not_matched (str | None)
• when_not_matched_by_source (str | None)

when_matched pydantic-field

when_matched = None

Replaces the derived 'row changed' condition on matched updates.

when_not_matched pydantic-field

when_not_matched = None

Extra condition applied when inserting unmatched source rows.

when_not_matched_by_source pydantic-field

when_not_matched_by_source = None

Extra condition applied to target rows missing from the source. Requires missing_in_source='delete'.

kelp.models.model.Column pydantic-model

Bases: BaseModel

Column definition for a model.

Show JSON schema:

{
  "$defs": {
    "GeneratedExpressionColumnConfig": {
      "properties": {
        "type": {
          "const": "expression",
          "description": "Column type identifier",
          "title": "Type",
          "type": "string"
        },
        "expression": {
          "description": "SQL expression used to generate the column value",
          "title": "Expression",
          "type": "string"
        }
      },
      "required": [
        "type",
        "expression"
      ],
      "title": "GeneratedExpressionColumnConfig",
      "type": "object"
    },
    "GeneratedIdentityColumnConfig": {
      "properties": {
        "type": {
          "const": "identity",
          "description": "Column type identifier",
          "title": "Type",
          "type": "string"
        },
        "as_default": {
          "default": false,
          "description": "Generated as default (True) or always (False)",
          "title": "As Default",
          "type": "boolean"
        },
        "start_with": {
          "default": 1,
          "description": "Starting value for the identity sequence",
          "title": "Start With",
          "type": "integer"
        },
        "increment_by": {
          "default": 1,
          "description": "Increment step for the identity sequence",
          "title": "Increment By",
          "type": "integer"
        }
      },
      "required": [
        "type"
      ],
      "title": "GeneratedIdentityColumnConfig",
      "type": "object"
    }
  },
  "description": "Column definition for a model.",
  "properties": {
    "name": {
      "description": "Column name",
      "title": "Name",
      "type": "string"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Human-readable description of the column",
      "title": "Description"
    },
    "data_type": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "SQL data type of the column",
      "title": "Data Type"
    },
    "nullable": {
      "default": true,
      "description": "Whether the column allows NULL values",
      "title": "Nullable",
      "type": "boolean"
    },
    "generated": {
      "anyOf": [
        {
          "discriminator": {
            "mapping": {
              "expression": "#/$defs/GeneratedExpressionColumnConfig",
              "identity": "#/$defs/GeneratedIdentityColumnConfig"
            },
            "propertyName": "type"
          },
          "oneOf": [
            {
              "$ref": "#/$defs/GeneratedIdentityColumnConfig"
            },
            {
              "$ref": "#/$defs/GeneratedExpressionColumnConfig"
            }
          ]
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Configuration for generated columns (identity or expression based)",
      "title": "Generated"
    },
    "tags": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "Metadata tags for the column",
      "title": "Tags",
      "type": "object"
    },
    "meta": {
      "additionalProperties": true,
      "description": "Generic user-defined metadata for filtering and grouping columns",
      "title": "Meta",
      "type": "object"
    }
  },
  "required": [
    "name"
  ],
  "title": "Column",
  "type": "object"
}

Fields:

• name (str)
• description (str | None)
• data_type (str | None)
• nullable (bool)
• generated (GeneratedIdentityColumnConfig | GeneratedExpressionColumnConfig | None)
• tags (dict[str, str])
• meta (dict[str, Any])

name pydantic-field

name

Column name

description pydantic-field

description = None

Human-readable description of the column

data_type pydantic-field

data_type = None

SQL data type of the column

nullable pydantic-field

nullable = True

Whether the column allows NULL values

generated pydantic-field

generated = None

Configuration for generated columns (identity or expression based)

tags pydantic-field

tags

Metadata tags for the column

meta pydantic-field

meta

Generic user-defined metadata for filtering and grouping columns

kelp.models.model.GeneratedIdentityColumnConfig pydantic-model

Bases: BaseModel

Show JSON schema:

{
  "properties": {
    "type": {
      "const": "identity",
      "description": "Column type identifier",
      "title": "Type",
      "type": "string"
    },
    "as_default": {
      "default": false,
      "description": "Generated as default (True) or always (False)",
      "title": "As Default",
      "type": "boolean"
    },
    "start_with": {
      "default": 1,
      "description": "Starting value for the identity sequence",
      "title": "Start With",
      "type": "integer"
    },
    "increment_by": {
      "default": 1,
      "description": "Increment step for the identity sequence",
      "title": "Increment By",
      "type": "integer"
    }
  },
  "required": [
    "type"
  ],
  "title": "GeneratedIdentityColumnConfig",
  "type": "object"
}

Fields:

• type (Literal['identity'])
• as_default (bool)
• start_with (int)
• increment_by (int)

type pydantic-field

type

Column type identifier

as_default pydantic-field

as_default = False

Generated as default (True) or always (False)

start_with pydantic-field

start_with = 1

Starting value for the identity sequence

increment_by pydantic-field

increment_by = 1

Increment step for the identity sequence

kelp.models.model.GeneratedExpressionColumnConfig pydantic-model

Bases: BaseModel

Show JSON schema:

{
  "properties": {
    "type": {
      "const": "expression",
      "description": "Column type identifier",
      "title": "Type",
      "type": "string"
    },
    "expression": {
      "description": "SQL expression used to generate the column value",
      "title": "Expression",
      "type": "string"
    }
  },
  "required": [
    "type",
    "expression"
  ],
  "title": "GeneratedExpressionColumnConfig",
  "type": "object"
}

Fields:

• type (Literal['expression'])
• expression (str)

type pydantic-field

type

Column type identifier

expression pydantic-field

expression

SQL expression used to generate the column value

kelp.models.model.Constraint pydantic-model

Bases: BaseModel

Show JSON schema:

{
  "properties": {
    "name": {
      "description": "Constraint name",
      "title": "Name",
      "type": "string"
    }
  },
  "required": [
    "name"
  ],
  "title": "Constraint",
  "type": "object"
}

Fields:

• name (str)

name pydantic-field

name

Constraint name

kelp.models.model.PrimaryKeyConstraint pydantic-model

Bases: Constraint

Show JSON schema:

{
  "properties": {
    "name": {
      "description": "Constraint name",
      "title": "Name",
      "type": "string"
    },
    "type": {
      "default": "primary_key",
      "description": "Constraint type identifier",
      "title": "Type",
      "type": "string"
    },
    "columns": {
      "description": "List of column names forming the primary key",
      "items": {
        "type": "string"
      },
      "title": "Columns",
      "type": "array"
    }
  },
  "required": [
    "name"
  ],
  "title": "PrimaryKeyConstraint",
  "type": "object"
}

Fields:

• name (str)
• type (str)
• columns (list[str])

type pydantic-field

type = 'primary_key'

Constraint type identifier

columns pydantic-field

columns

List of column names forming the primary key

name pydantic-field

name

Constraint name

kelp.models.model.ForeignKeyConstraint pydantic-model

Bases: Constraint

Show JSON schema:

{
  "properties": {
    "name": {
      "description": "Constraint name",
      "title": "Name",
      "type": "string"
    },
    "type": {
      "default": "foreign_key",
      "description": "Constraint type identifier",
      "title": "Type",
      "type": "string"
    },
    "columns": {
      "description": "List of local column names",
      "items": {
        "type": "string"
      },
      "title": "Columns",
      "type": "array"
    },
    "reference_table": {
      "description": "Fully qualified name of the referenced table",
      "title": "Reference Table",
      "type": "string"
    },
    "reference_columns": {
      "description": "List of column names in the referenced table",
      "items": {
        "type": "string"
      },
      "title": "Reference Columns",
      "type": "array"
    }
  },
  "required": [
    "name",
    "reference_table"
  ],
  "title": "ForeignKeyConstraint",
  "type": "object"
}

Fields:

• name (str)
• type (str)
• columns (list[str])
• reference_table (str)
• reference_columns (list[str])

type pydantic-field

type = 'foreign_key'

Constraint type identifier

columns pydantic-field

columns

List of local column names

reference_table pydantic-field

reference_table

Fully qualified name of the referenced table

reference_columns pydantic-field

reference_columns

List of column names in the referenced table

name pydantic-field

name

Constraint name

kelp.models.model.Quality pydantic-model

Bases: BaseModel

Show JSON schema:

{
  "properties": {
    "engine": {
      "description": "Quality engine type",
      "title": "Engine",
      "type": "string"
    },
    "level": {
      "default": "row",
      "description": "Level at which quality is enforced",
      "enum": [
        "row",
        "table"
      ],
      "title": "Level",
      "type": "string"
    }
  },
  "required": [
    "engine"
  ],
  "title": "Quality",
  "type": "object"
}

Fields:

• engine (str)
• level (Literal['row', 'table'])

engine pydantic-field

engine

Quality engine type

level pydantic-field

level = 'row'

Level at which quality is enforced

kelp.models.model.SDPQuality pydantic-model

Bases: Quality

Show JSON schema:

{
  "properties": {
    "engine": {
      "const": "sdp",
      "description": "Quality engine type",
      "title": "Engine",
      "type": "string"
    },
    "level": {
      "const": "row",
      "default": "row",
      "description": "Quality enforcement level",
      "title": "Level",
      "type": "string"
    },
    "expect_all": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "SQL expressions that must pass",
      "title": "Expect All",
      "type": "object"
    },
    "expect_all_or_drop": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "SQL expressions; failing rows are dropped",
      "title": "Expect All Or Drop",
      "type": "object"
    },
    "expect_all_or_fail": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "SQL expressions; job fails if any expression fails",
      "title": "Expect All Or Fail",
      "type": "object"
    },
    "expect_all_or_quarantine": {
      "additionalProperties": {
        "type": "string"
      },
      "description": "SQL expressions; failing rows are quarantined",
      "title": "Expect All Or Quarantine",
      "type": "object"
    }
  },
  "required": [
    "engine"
  ],
  "title": "SDPQuality",
  "type": "object"
}

Fields:

• engine (Literal['sdp'])
• level (Literal['row'])
• expect_all (dict[str, str])
• expect_all_or_drop (dict[str, str])
• expect_all_or_fail (dict[str, str])
• expect_all_or_quarantine (dict[str, str])

engine pydantic-field

engine

Quality engine type

level pydantic-field

level = 'row'

Quality enforcement level

expect_all pydantic-field

expect_all

SQL expressions that must pass

expect_all_or_drop pydantic-field

expect_all_or_drop

SQL expressions; failing rows are dropped

expect_all_or_fail pydantic-field

expect_all_or_fail

SQL expressions; job fails if any expression fails

expect_all_or_quarantine pydantic-field

expect_all_or_quarantine

SQL expressions; failing rows are quarantined

kelp.models.model.DQXQuality pydantic-model

Bases: Quality

Show JSON schema:

{
  "properties": {
    "engine": {
      "const": "dqx",
      "description": "Quality engine type",
      "title": "Engine",
      "type": "string"
    },
    "level": {
      "default": "row",
      "description": "Level at which quality is enforced",
      "enum": [
        "row",
        "table"
      ],
      "title": "Level",
      "type": "string"
    },
    "sdp_expect_level": {
      "default": "warn",
      "description": "Action for quality violations: warn, fail, drop, or deactivate",
      "enum": [
        "warn",
        "fail",
        "drop",
        "deactivate"
      ],
      "title": "Sdp Expect Level",
      "type": "string"
    },
    "sdp_quarantine": {
      "default": false,
      "description": "Whether to quarantine rows failing quality checks",
      "title": "Sdp Quarantine",
      "type": "boolean"
    },
    "spark_violation_action": {
      "default": "error",
      "description": "Action for quality violations at the Spark level: error (raise exception), ignore (store errors in output), or drop (exclude from output)",
      "enum": [
        "error",
        "ignore",
        "drop"
      ],
      "title": "Spark Violation Action",
      "type": "string"
    },
    "spark_quarantine": {
      "default": false,
      "description": "Whether to write rows failing quality checks to a quarantine table at the Spark level",
      "title": "Spark Quarantine",
      "type": "boolean"
    },
    "checks": {
      "description": "Quality check configurations",
      "items": {
        "additionalProperties": true,
        "type": "object"
      },
      "title": "Checks",
      "type": "array"
    }
  },
  "required": [
    "engine"
  ],
  "title": "DQXQuality",
  "type": "object"
}

Fields:

• level (Literal['row', 'table'])
• engine (Literal['dqx'])
• sdp_expect_level (Literal['warn', 'fail', 'drop', 'deactivate'])
• sdp_quarantine (bool)
• spark_violation_action (Literal['error', 'ignore', 'drop'])
• spark_quarantine (bool)
• checks (list[dict])

engine pydantic-field

engine

Quality engine type

sdp_expect_level pydantic-field

sdp_expect_level = 'warn'

Action for quality violations: warn, fail, drop, or deactivate

sdp_quarantine pydantic-field

sdp_quarantine = False

Whether to quarantine rows failing quality checks

spark_violation_action pydantic-field

spark_violation_action = 'error'

Action for quality violations at the Spark level: error (raise exception), ignore (store errors in output), or drop (exclude from output)

spark_quarantine pydantic-field

spark_quarantine = False

Whether to write rows failing quality checks to a quarantine table at the Spark level

checks pydantic-field

checks

Quality check configurations

level pydantic-field

level = 'row'

Level at which quality is enforced