Tables

kelp.tables

Generic model metadata API for use in any Spark job.

REGISTRY module-attribute

REGISTRY = ModelRegistry()

FullRefreshStrategy module-attribute

FullRefreshStrategy = Literal['drop', 'replace']

MaterializationConfig module-attribute

MaterializationConfig = Annotated[
    AppendConfig
    | OverwriteConfig
    | MergeConfig
    | Scd2Config,
    Field(discriminator="mode"),
]

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'])

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.

mode pydantic-field

mode = 'append'

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 [])

MaterializationOptions pydantic-model

Bases: BaseModel

Switches for the steps kelp runs around a materialization write.

These are operational choices, not part of a model's write semantics, so they live outside the mode config: project-wide defaults come from kelp_project.yml, and a call site may override individual switches.

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Switches for the steps kelp runs around a materialization write.\n\nThese are operational choices, not part of a model's write semantics, so they\nlive outside the mode config: project-wide defaults come from\n``kelp_project.yml``, and a call site may override individual switches.",
  "properties": {
    "apply_quality_checks": {
      "default": true,
      "description": "Whether DQX checks declared in model metadata are applied.",
      "title": "Apply Quality Checks",
      "type": "boolean"
    },
    "sync_metadata": {
      "default": true,
      "description": "Whether catalog metadata is synced afterwards. Requires a model.",
      "title": "Sync Metadata",
      "type": "boolean"
    },
    "apply_optimize": {
      "default": true,
      "description": "Whether OPTIMIZE runs after the write.",
      "title": "Apply Optimize",
      "type": "boolean"
    },
    "apply_vacuum": {
      "default": true,
      "description": "Whether VACUUM runs after the write.",
      "title": "Apply Vacuum",
      "type": "boolean"
    },
    "vacuum_lite": {
      "default": true,
      "description": "Whether VACUUM uses LITE mode. Only used when apply_vacuum is set.",
      "title": "Vacuum Lite",
      "type": "boolean"
    }
  },
  "title": "MaterializationOptions",
  "type": "object"
}

Config:

• extra: forbid

Fields:

• apply_quality_checks (bool)
• sync_metadata (bool)
• apply_optimize (bool)
• apply_vacuum (bool)
• vacuum_lite (bool)

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

apply_quality_checks pydantic-field

apply_quality_checks = True

Whether DQX checks declared in model metadata are applied.

sync_metadata pydantic-field

sync_metadata = True

Whether catalog metadata is synced afterwards. Requires a model.

apply_optimize pydantic-field

apply_optimize = True

Whether OPTIMIZE runs after the write.

apply_vacuum pydantic-field

apply_vacuum = True

Whether VACUUM runs after the write.

vacuum_lite pydantic-field

vacuum_lite = True

Whether VACUUM uses LITE mode. Only used when apply_vacuum is set.

merged_with

merged_with(override)

Return these options with the explicitly set fields of override applied.

Parameters:

Name	Type	Description	Default
override	MaterializationOptions | dict | None	Options or mapping overriding individual switches, or None.	required

Returns:

Type	Description
MaterializationOptions	The effective options. Unset fields keep this instance's values, so a
MaterializationOptions	call site only has to state what it wants to change.

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

def merged_with(
    self, override: "MaterializationOptions | dict | None"
) -> "MaterializationOptions":
    """Return these options with the explicitly set fields of ``override`` applied.

    Args:
        override: Options or mapping overriding individual switches, or ``None``.

    Returns:
        The effective options. Unset fields keep this instance's values, so a
        call site only has to state what it wants to change.
    """
    if override is None:
        return self
    if isinstance(override, dict):
        return self.model_copy(update=override)
    return self.model_copy(update=override.model_dump(exclude_unset=True))

MaterializedContext dataclass

MaterializedContext(
    spark, this, target_exists, full_refresh=False
)

Execution context optionally injected into materialized functions.

Attributes:

Name	Type	Description
spark	SparkSession	Active SparkSession.
this	str	Fully qualified target table name (or provided name when unresolved).
target_exists	bool	Whether the target table exists before materialization.
full_refresh	bool	Whether a full refresh was requested by the caller.

spark instance-attribute

spark

this instance-attribute

this

target_exists instance-attribute

target_exists

full_refresh class-attribute instance-attribute

full_refresh = False

is_incremental

is_incremental()

Return True when target exists and full refresh is not requested.

Source code in src/kelp/tables/materialization/decorator.py

def is_incremental(self) -> bool:
    """Return ``True`` when target exists and full refresh is not requested."""
    return self.target_exists and not self.full_refresh

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'])

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.

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.

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)

ModelRegistry

ModelRegistry()

Registered materialized models and their dependency graph.

Create an empty registry.

Source code in src/kelp/tables/materialization/runner.py

def __init__(self) -> None:
    """Create an empty registry."""
    self._specs: dict[str, ModelSpec] = {}

register

register(spec)

Register a model, logging a warning when it replaces an existing name.

Parameters:

Name	Type	Description	Default
spec	ModelSpec	Model specification to register.	required

Source code in src/kelp/tables/materialization/runner.py

def register(self, spec: ModelSpec) -> None:
    """Register a model, logging a warning when it replaces an existing name.

    Args:
        spec: Model specification to register.
    """
    if spec.name in self._specs:
        logger.warning(
            "Model '%s' is already registered; replacing the previous definition",
            spec.name,
        )
    self._specs[spec.name] = spec

get

get(name)

Return the spec registered under name.

Parameters:

Name	Type	Description	Default
name	str	Model name.	required

Returns:

Type	Description
ModelSpec	The registered model specification.

Raises:

Type	Description
KeyError	If no model is registered under name.

Source code in src/kelp/tables/materialization/runner.py

def get(self, name: str) -> ModelSpec:
    """Return the spec registered under ``name``.

    Args:
        name: Model name.

    Returns:
        The registered model specification.

    Raises:
        KeyError: If no model is registered under ``name``.
    """
    return self._require(name)

names

names()

Return all registered model names in registration order.

Source code in src/kelp/tables/materialization/runner.py

def names(self) -> list[str]:
    """Return all registered model names in registration order."""
    return list(self._specs)

clear

clear()

Remove all registered models.

Source code in src/kelp/tables/materialization/runner.py

def clear(self) -> None:
    """Remove all registered models."""
    self._specs.clear()

toposort

toposort(names=None)

Dependency-ordered model names; all of them when names is None.

Transitive dependencies of the requested models are always included.

Parameters:

Name	Type	Description	Default
names	list[str] | None	Models to order, or None for every registered model.	None

Returns:

Type	Description
list[str]	Model names ordered so that every dependency precedes its consumers.

Raises:

Type	Description
KeyError	If a requested model or one of its dependencies is unknown.
ValueError	If the dependency graph contains a cycle.

Source code in src/kelp/tables/materialization/runner.py

def toposort(self, names: list[str] | None = None) -> list[str]:
    """Dependency-ordered model names; all of them when names is None.

    Transitive dependencies of the requested models are always included.

    Args:
        names: Models to order, or None for every registered model.

    Returns:
        Model names ordered so that every dependency precedes its consumers.

    Raises:
        KeyError: If a requested model or one of its dependencies is unknown.
        ValueError: If the dependency graph contains a cycle.
    """
    visited: set[str] = set()
    visiting: set[str] = set()
    order: list[str] = []

    def visit(name: str, path: list[str], required_by: str | None) -> None:
        if name in visited:
            return

        if name in visiting:
            cycle = " -> ".join([*path, name])
            raise ValueError(f"Cyclic dependency detected: {cycle}")

        spec = self._require(name, required_by=required_by)

        visiting.add(name)
        for dep in spec.depends_on:
            visit(dep, [*path, name], required_by=name)
        visiting.remove(name)

        visited.add(name)
        order.append(name)

    for name in self.names() if names is None else names:
        visit(name, [], None)

    return order

levels

levels(names=None)

Dependency levels: every model in a level may run concurrently.

Parameters:

Name	Type	Description	Default
names	list[str] | None	Models to group, or None for every registered model.	None

Returns:

Type	Description
list[list[str]]	Lists of model names, each level depending only on earlier levels.

Raises:

Type	Description
KeyError	If a requested model or one of its dependencies is unknown.
ValueError	If the dependency graph contains a cycle.

Source code in src/kelp/tables/materialization/runner.py

def levels(self, names: list[str] | None = None) -> list[list[str]]:
    """Dependency levels: every model in a level may run concurrently.

    Args:
        names: Models to group, or None for every registered model.

    Returns:
        Lists of model names, each level depending only on earlier levels.

    Raises:
        KeyError: If a requested model or one of its dependencies is unknown.
        ValueError: If the dependency graph contains a cycle.
    """
    order = self.toposort(names)
    selected = set(order)

    depth: dict[str, int] = {}
    for name in order:
        deps = [d for d in self._specs[name].depends_on if d in selected]
        depth[name] = 1 + max((depth[d] for d in deps), default=-1)

    levels: list[list[str]] = [[] for _ in range(max(depth.values(), default=-1) + 1)]
    for name in order:
        levels[depth[name]].append(name)

    return levels

ModelSpec dataclass

ModelSpec(name, fn, depends_on=list())

A registered materialized model.

Attributes:

Name	Type	Description
name	str	Unique model name.
fn	Callable[..., Any]	Callable executing the materialization, invoked with full_refresh and spark. The session is handed over rather than discovered by the callable, because PySpark's active session is thread-local and would be invisible to a model running on a worker thread.
depends_on	list[str]	Names of models that must run before this one.

name instance-attribute

name

fn instance-attribute

fn

depends_on class-attribute instance-attribute

depends_on = field(default_factory=list)

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)

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.

mode pydantic-field

mode = 'overwrite'

replace_where pydantic-field

replace_where = None

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

Runner

Runner(registry=None, spark=None)

Executes registered materialized models in dependency order.

Create a runner.

Parameters:

Name	Type	Description	Default
registry	ModelRegistry | None	Registry to run models from; defaults to the module-level REGISTRY.	None
spark	SparkSession | None	SparkSession handed to every model. Defaults to the session active when a run starts.	None

Source code in src/kelp/tables/materialization/runner.py

def __init__(
    self,
    registry: ModelRegistry | None = None,
    spark: SparkSession | None = None,
) -> None:
    """Create a runner.

    Args:
        registry: Registry to run models from; defaults to the module-level
            ``REGISTRY``.
        spark: SparkSession handed to every model. Defaults to the session
            active when a run starts.
    """
    self.registry = registry if registry is not None else REGISTRY
    self.runlog = RunLog()
    self._spark = spark
    self._runlog_lock = threading.Lock()

registry instance-attribute

registry = registry if registry is not None else REGISTRY

runlog instance-attribute

runlog = RunLog()

plan_one

plan_one(name)

Dependency-ordered names needed to build name, including name.

Parameters:

Name	Type	Description	Default
name	str	Model to plan.	required

Returns:

Type	Description
list[str]	Model names in the order they must run.

Source code in src/kelp/tables/materialization/runner.py

def plan_one(self, name: str) -> list[str]:
    """Dependency-ordered names needed to build ``name``, including ``name``.

    Args:
        name: Model to plan.

    Returns:
        Model names in the order they must run.
    """
    return self.registry.toposort([name])

plan_all

plan_all()

Dependency-ordered names of every registered model.

Source code in src/kelp/tables/materialization/runner.py

def plan_all(self) -> list[str]:
    """Dependency-ordered names of every registered model."""
    return self.registry.toposort()

run_one

run_one(name, full_refresh=False)

Run exactly this one model and return its result.

Upstreams are NOT run - use run(plan_one(name)) for that.

Parameters:

Name	Type	Description	Default
name	str	Model to run.	required
full_refresh	bool	Whether to rebuild the target from scratch.	False

Returns:

Type	Description
Any	Whatever the model function returns.

Source code in src/kelp/tables/materialization/runner.py

def run_one(self, name: str, full_refresh: bool = False) -> Any:
    """Run exactly this one model and return its result.

    Upstreams are NOT run - use ``run(plan_one(name))`` for that.

    Args:
        name: Model to run.
        full_refresh: Whether to rebuild the target from scratch.

    Returns:
        Whatever the model function returns.
    """
    self.registry.get(name)
    return self._run_model(name, full_refresh=full_refresh, spark=self._resolve_spark())

run

run(
    names=None,
    full_refresh=False,
    parallel=False,
    max_workers=4,
)

Run models in dependency order, optionally running each level concurrently.

With parallel=True the models of a dependency level are submitted from several threads to the same SparkSession, which Spark supports. The session is resolved once here and handed to each model, since worker threads cannot see the active session themselves.

Parameters:

Name	Type	Description	Default
names	list[str] | None	Models to run, or None for every registered model.	None
full_refresh	bool	Whether to rebuild the targets from scratch.	False
parallel	bool	Whether to run independent models of a level concurrently.	False
max_workers	int	Maximum number of threads used when parallel is True.	4

Source code in src/kelp/tables/materialization/runner.py

def run(
    self,
    names: list[str] | None = None,
    full_refresh: bool = False,
    parallel: bool = False,
    max_workers: int = 4,
) -> None:
    """Run models in dependency order, optionally running each level concurrently.

    With ``parallel=True`` the models of a dependency level are submitted from
    several threads to the same SparkSession, which Spark supports. The session
    is resolved once here and handed to each model, since worker threads cannot
    see the active session themselves.

    Args:
        names: Models to run, or None for every registered model.
        full_refresh: Whether to rebuild the targets from scratch.
        parallel: Whether to run independent models of a level concurrently.
        max_workers: Maximum number of threads used when ``parallel`` is True.
    """
    spark = self._resolve_spark()

    if not parallel:
        for name in self.registry.toposort(names):
            self._run_model(name, full_refresh=full_refresh, spark=spark)
        return

    for level in self.registry.levels(names):
        if len(level) == 1:
            self._run_model(level[0], full_refresh=full_refresh, spark=spark)
            continue

        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures = [
                pool.submit(self._run_model, name, full_refresh=full_refresh, spark=spark)
                for name in level
            ]
            errors = [future.exception() for future in futures]

        for error in errors:
            if error is not None:
                raise error

run_all

run_all(full_refresh=False, parallel=False, max_workers=4)

Run every registered model in dependency order.

Parameters:

Name	Type	Description	Default
full_refresh	bool	Whether to rebuild the targets from scratch.	False
parallel	bool	Whether to run independent models of a level concurrently.	False
max_workers	int	Maximum number of threads used when parallel is True.	4

Source code in src/kelp/tables/materialization/runner.py

def run_all(
    self,
    full_refresh: bool = False,
    parallel: bool = False,
    max_workers: int = 4,
) -> None:
    """Run every registered model in dependency order.

    Args:
        full_refresh: Whether to rebuild the targets from scratch.
        parallel: Whether to run independent models of a level concurrently.
        max_workers: Maximum number of threads used when ``parallel`` is True.
    """
    self.run(None, full_refresh=full_refresh, parallel=parallel, max_workers=max_workers)

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

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)

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.

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.

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)

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'.

init

init(
    project_file_path=None,
    target=None,
    init_vars=None,
    manifest_file_path=None,
    refresh=False,
    store_in_global=True,
    run_policy_checks=False,
    log_level=None,
)

Initialize kelp runtime context from current directory.

When manifest_file_path is provided (or resolved from KELP_MANIFEST_FILE environment variable), the context is loaded directly from a pre-built manifest JSON file, skipping all project discovery, Jinja rendering, and metadata loading.

When policy_config.enabled is True in the project settings, metadata governance policies are evaluated immediately after loading. Warn-severity violations are logged; error-severity violations raise a RuntimeError.

Parameters:

Name	Type	Description	Default
project_file_path	str | None	Path to project file or directory.	None
target	str | None	Target environment name.	None
init_vars	dict[str, Any] | None	Runtime variable overrides.	None
manifest_file_path	str | None	Path to a manifest JSON file. When provided, skips source file loading. Also resolved from KELP_MANIFEST_FILE env var.	None
refresh	bool	If True, recreate context even if one already exists.	False
store_in_global	bool	Whether to store context globally.	True
run_policy_checks	bool	Whether to run policy checks.	False
log_level	str | None	Optional log level to configure.	None

Returns:

Type	Description
MetaRuntimeContext	The initialized MetaRuntimeContext.

Source code in src/kelp/config/config.py

def init(
    project_file_path: str | None = None,
    target: str | None = None,
    init_vars: dict[str, Any] | None = None,
    manifest_file_path: str | None = None,
    refresh: bool = False,
    store_in_global: bool = True,
    run_policy_checks: bool = False,
    log_level: str | None = None,
) -> MetaRuntimeContext:
    """Initialize kelp runtime context from current directory.

    When ``manifest_file_path`` is provided (or resolved from ``KELP_MANIFEST_FILE``
    environment variable), the context is loaded directly from a pre-built
    manifest JSON file, skipping all project discovery, Jinja rendering, and
    metadata loading.

    When ``policy_config.enabled`` is True in the project settings, metadata
    governance policies are evaluated immediately after loading. Warn-severity
    violations are logged; error-severity violations raise a RuntimeError.

    Args:
        project_file_path: Path to project file or directory.
        target: Target environment name.
        init_vars: Runtime variable overrides.
        manifest_file_path: Path to a manifest JSON file. When provided, skips
            source file loading. Also resolved from KELP_MANIFEST_FILE env var.
        refresh: If True, recreate context even if one already exists.
        store_in_global: Whether to store context globally.
        run_policy_checks: Whether to run policy checks.
        log_level: Optional log level to configure.

    Returns:
        The initialized MetaRuntimeContext.
    """
    if log_level:
        configure_logging(log_level)

    ctx = KelpFramework.init(
        project_file_path=project_file_path,
        target=target,
        init_vars=init_vars,
        manifest_file_path=manifest_file_path,
        refresh=refresh,
        store_in_global=store_in_global,
    )

    _run_policy_checks(ctx, run_policy_checks)

    return ctx

columns

columns(name)

Get the column definitions for a model.

Source code in src/kelp/tables/api.py

def columns(name: str) -> list[Column]:
    """Get the column definitions for a model."""
    model = ModelManager.build_model(name)
    if model.root_model:
        return model.root_model.columns
    return []

ddl

ddl(name, if_not_exists=True)

Get the full CREATE TABLE DDL statement for a model.

Source code in src/kelp/tables/api.py

def ddl(name: str, if_not_exists: bool = True) -> str | None:
    """Get the full CREATE TABLE DDL statement for a model."""
    return ModelManager.build_model(name).get_ddl(if_not_exists=if_not_exists)

func

func(name)

Get the fully qualified name for a Unity Catalog function.

Source code in src/kelp/tables/api.py

def func(name: str) -> str:
    """Get the fully qualified name for a Unity Catalog function."""
    from kelp.config import get_context

    context = get_context()
    return context.catalog_index.get("functions", name).get_qualified_name()

get_model

get_model(name)

Get the KelpModel object for a given model name.

Source code in src/kelp/tables/api.py

def get_model(name: str) -> KelpModel:
    """Get the KelpModel object for a given model name."""
    return ModelManager.build_model(name)

ref

ref(name)

Get the fully qualified name for a model.

Source code in src/kelp/tables/api.py

def ref(name: str) -> str:
    """Get the fully qualified name for a model."""
    return ModelManager.build_model(name).fqn or name

schema

schema(name, exclude=None)

Get the Spark schema DDL for a model.

Parameters:

Name	Type	Description	Default
name	str	Model name.	required
exclude	list[str] | None	Column names to exclude from the schema.	None

Returns:

Type	Description
str | None	Spark schema DDL string, or None if not available.

Source code in src/kelp/tables/api.py

def schema(name: str, exclude: list[str] | None = None) -> str | None:
    """Get the Spark schema DDL for a model.

    Args:
        name: Model name.
        exclude: Column names to exclude from the schema.

    Returns:
        Spark schema DDL string, or ``None`` if not available.
    """
    return ModelManager.build_model(name, exclude=exclude).schema

schema_lite

schema_lite(name, exclude=None)

Get the raw Spark schema without constraints or generated columns.

Parameters:

Name	Type	Description	Default
name	str	Model name.	required
exclude	list[str] | None	Column names to exclude from the schema.	None

Returns:

Type	Description
str | None	Spark schema DDL string, or None if not available.

Source code in src/kelp/tables/api.py

def schema_lite(name: str, exclude: list[str] | None = None) -> str | None:
    """Get the raw Spark schema without constraints or generated columns.

    Args:
        name: Model name.
        exclude: Column names to exclude from the schema.

    Returns:
        Spark schema DDL string, or ``None`` if not available.
    """
    return ModelManager.build_model(name, exclude=exclude).schema_lite

source

source(name)

Get the path for a data source.

Source code in src/kelp/tables/api.py

def source(name: str) -> str:
    """Get the path for a data source."""
    from kelp.service.source_manager import SourceManager

    return SourceManager.get_path(name)

source_options

source_options(name)

Get the options dictionary for a data source.

Source code in src/kelp/tables/api.py

def source_options(name: str) -> dict:
    """Get the options dictionary for a data source."""
    from kelp.service.source_manager import SourceManager

    return SourceManager.get_options(name)

materialize

materialize(
    *,
    dataframe,
    name,
    config=None,
    options=None,
    full_refresh=False,
    full_refresh_strategy="drop",
    spark=None,
)

Materialize a DataFrame to Delta Lake according to a materialization config.

Parameters:

Name	Type	Description	Default
dataframe	DataFrame	DataFrame to materialize.	required
name	str	Unqualified kelp model name, or a qualified table name to write without metadata.	required
config	MaterializationConfig | dict | None	Materialization config or mapping. Replaces the model's config entirely when both are present.	None
options	MaterializationOptions | dict | None	Overrides for the steps run around the write — quality checks, catalog sync, OPTIMIZE and VACUUM. Unset switches fall back to the project's materialization_options.	None
full_refresh	bool	Whether to rebuild the target from scratch first. Ignored with a warning when the config sets allow_full_refresh: false.	False
full_refresh_strategy	FullRefreshStrategy	How a full refresh resets the target — drop to drop and recreate it, replace to keep the table (and its grants and history) via CREATE OR REPLACE TABLE, or TRUNCATE when no DDL is available.	'drop'
spark	SparkSession | None	SparkSession to use. Defaults to the active session.	None

Returns:

Type	Description
DataFrame	The DataFrame that was written, which is the input frame minus any rows
DataFrame	quality checks dropped.

Raises:

Type	Description
RuntimeError	If no SparkSession is available.
LookupError	If name is unqualified and resolves to no kelp model.
ValueError	If quality checks found errors and the model's spark_violation_action is error.

Source code in src/kelp/tables/materialization/orchestrator.py

def materialize(
    *,
    dataframe: DataFrame,
    name: str,
    config: "MaterializationConfig | dict | None" = None,
    options: "MaterializationOptions | dict | None" = None,
    full_refresh: bool = False,
    full_refresh_strategy: FullRefreshStrategy = "drop",
    spark: SparkSession | None = None,
) -> DataFrame:
    """Materialize a DataFrame to Delta Lake according to a materialization config.

    Args:
        dataframe: DataFrame to materialize.
        name: Unqualified kelp model name, or a qualified table name to write
            without metadata.
        config: Materialization config or mapping. Replaces the model's config
            entirely when both are present.
        options: Overrides for the steps run around the write — quality checks,
            catalog sync, OPTIMIZE and VACUUM. Unset switches fall back to the
            project's ``materialization_options``.
        full_refresh: Whether to rebuild the target from scratch first. Ignored with
            a warning when the config sets ``allow_full_refresh: false``.
        full_refresh_strategy: How a full refresh resets the target — ``drop`` to
            drop and recreate it, ``replace`` to keep the table (and its grants and
            history) via ``CREATE OR REPLACE TABLE``, or ``TRUNCATE`` when no DDL
            is available.
        spark: SparkSession to use. Defaults to the active session.

    Returns:
        The DataFrame that was written, which is the input frame minus any rows
        quality checks dropped.

    Raises:
        RuntimeError: If no SparkSession is available.
        LookupError: If ``name`` is unqualified and resolves to no kelp model.
        ValueError: If quality checks found errors and the model's
            ``spark_violation_action`` is ``error``.
    """
    resolved = resolve_materialization_inputs(table_name=name, config=config)
    return materialize_resolved(
        dataframe=dataframe,
        resolved=resolved,
        options=options,
        full_refresh=full_refresh,
        full_refresh_strategy=full_refresh_strategy,
        spark=spark,
    )

materialized

materialized(
    func: Callable[..., DataFrame],
) -> Callable[..., DataFrame]

materialized(
    *,
    name: str | None = None,
    config: MaterializationConfig | dict | None = None,
    options: MaterializationOptions | dict | None = None,
    depends_on: list[str] | None = None,
    full_refresh: bool = False,
    full_refresh_strategy: FullRefreshStrategy = "drop",
) -> Callable[
    [Callable[..., DataFrame]], Callable[..., DataFrame]
]

materialized(
    func=None,
    *,
    name=None,
    config=None,
    options=None,
    depends_on=None,
    full_refresh=False,
    full_refresh_strategy="drop",
)

Decorator that materializes the returned DataFrame.

Usable bare (@materialized) or called (@materialized(name=...)). Model matching uses name when provided; otherwise the wrapped function name is used. An unqualified name must match a kelp model; pass a qualified table name to materialize without metadata.

The wrapper accepts full_refresh and spark keywords at call time, which override the decorator's value and the active session and are not passed on to the wrapped function.

Parameters:

Name	Type	Description	Default
func	Callable[..., DataFrame] | None	The decorated function when used bare.	None
name	str | None	Optional kelp model name, or a qualified table name.	None
config	MaterializationConfig | dict | None	Optional materialization config, replacing the model's config.	None
options	MaterializationOptions | dict | None	Overrides for the steps run around the write — quality checks, catalog sync, OPTIMIZE and VACUUM. Unset switches fall back to the project's materialization_options.	None
depends_on	list[str] | None	Model names this model must run after, for the runner.	None
full_refresh	bool	Whether to rebuild the target from scratch before writing.	False
full_refresh_strategy	FullRefreshStrategy	How a full refresh resets the target — drop to drop and recreate it, replace to keep the table (and its grants and history) in place.	'drop'

Returns:

Type	Description
Callable[..., DataFrame] | Callable[[Callable[..., DataFrame]], Callable[..., DataFrame]]	The decorated callable, or the decorator when called with options.

Source code in src/kelp/tables/materialization/decorator.py

def materialized(
    func: Callable[..., DataFrame] | None = None,
    *,
    name: str | None = None,
    config: MaterializationConfig | dict | None = None,
    options: MaterializationOptions | dict | None = None,
    depends_on: list[str] | None = None,
    full_refresh: bool = False,
    full_refresh_strategy: FullRefreshStrategy = "drop",
) -> Callable[..., DataFrame] | Callable[[Callable[..., DataFrame]], Callable[..., DataFrame]]:
    """Decorator that materializes the returned DataFrame.

    Usable bare (``@materialized``) or called (``@materialized(name=...)``).
    Model matching uses `name` when provided; otherwise the wrapped function
    name is used. An unqualified name must match a kelp model; pass a qualified
    table name to materialize without metadata.

    The wrapper accepts ``full_refresh`` and ``spark`` keywords at call time, which
    override the decorator's value and the active session and are not passed on to
    the wrapped function.

    Args:
        func: The decorated function when used bare.
        name: Optional kelp model name, or a qualified table name.
        config: Optional materialization config, replacing the model's config.
        options: Overrides for the steps run around the write — quality checks,
            catalog sync, OPTIMIZE and VACUUM. Unset switches fall back to the
            project's ``materialization_options``.
        depends_on: Model names this model must run after, for the runner.
        full_refresh: Whether to rebuild the target from scratch before writing.
        full_refresh_strategy: How a full refresh resets the target — ``drop`` to
            drop and recreate it, ``replace`` to keep the table (and its grants and
            history) in place.

    Returns:
        The decorated callable, or the decorator when called with options.
    """
    cfg = parse_materialization_config(config)
    depends_on = depends_on or []

    def decorator(fn: Callable[..., DataFrame]) -> Callable[..., DataFrame]:
        function_name = getattr(fn, "__name__", fn.__class__.__name__)
        target_name = name or function_name

        signature = inspect.signature(fn)
        parameters = list(signature.parameters.values())
        inject_ctx = bool(
            parameters
            and parameters[0].kind
            in {
                inspect.Parameter.POSITIONAL_ONLY,
                inspect.Parameter.POSITIONAL_OR_KEYWORD,
            }
            and (
                parameters[0].name in {"ctx", "context"}
                or parameters[0].annotation is MaterializedContext
            )
        )

        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> DataFrame:
            runtime_full_refresh = kwargs.pop("full_refresh", full_refresh)

            # A caller may hand the session over: PySpark's active session is
            # thread-local, so a model run by the Runner in a worker thread cannot
            # find it itself.
            spark = kwargs.pop("spark", None) or SparkSession.getActiveSession()
            if spark is None:
                raise RuntimeError("No active SparkSession available for materialization.")

            # Resolved once here and handed to the orchestrator, so metadata is not
            # looked up twice for the same run.
            resolved = resolve_materialization_inputs(table_name=target_name, config=cfg)

            call_args = args
            if inject_ctx:
                context = MaterializedContext(
                    spark=spark,
                    this=resolved.target_name,
                    target_exists=table_exists(spark, resolved.target_name),
                    full_refresh=runtime_full_refresh,
                )
                call_args = (context, *args)

            result = fn(*call_args, **kwargs)
            if not isinstance(result, DataFrame):
                raise TypeError(
                    f"Materialized function '{target_name}' must return DataFrame, "
                    f"got {type(result).__name__}."
                )

            return materialize_resolved(
                dataframe=result,
                resolved=resolved,
                options=options,
                full_refresh=runtime_full_refresh,
                full_refresh_strategy=full_refresh_strategy,
                spark=spark,
            )

        REGISTRY.register(
            ModelSpec(
                name=target_name,
                fn=wrapper,
                depends_on=list(depends_on),
            )
        )
        return wrapper

    if func is not None and callable(func):
        return decorator(func)

    return decorator

kelp.service.model_manager.KelpModel dataclass

KelpModel(
    name,
    table_type=None,
    comment=None,
    table_properties=None,
    spark_conf=None,
    path=None,
    partition_cols=None,
    cluster_by_auto=None,
    cluster_by=None,
    row_filter=None,
    auto_ttl=None,
    fqn=None,
    schema=None,
    schema_lite=None,
    dqx_quality=None,
    validation_table=None,
    quarantine_table=None,
    target_table=None,
    root_model=None,
    materialization=None,
    meta=None,
)

name instance-attribute

name

table_type class-attribute instance-attribute

table_type = None

comment class-attribute instance-attribute

comment = None

table_properties class-attribute instance-attribute

table_properties = None

spark_conf class-attribute instance-attribute

spark_conf = None

path class-attribute instance-attribute

path = None

partition_cols class-attribute instance-attribute

partition_cols = None

cluster_by_auto class-attribute instance-attribute

cluster_by_auto = None

cluster_by class-attribute instance-attribute

cluster_by = None

row_filter class-attribute instance-attribute

row_filter = None

auto_ttl class-attribute instance-attribute

auto_ttl = None

fqn class-attribute instance-attribute

fqn = None

schema class-attribute instance-attribute

schema = None

schema_lite class-attribute instance-attribute

schema_lite = None

dqx_quality class-attribute instance-attribute

dqx_quality = None

validation_table class-attribute instance-attribute

validation_table = None

quarantine_table class-attribute instance-attribute

quarantine_table = None

target_table class-attribute instance-attribute

target_table = None

root_model class-attribute instance-attribute

root_model = None

materialization class-attribute instance-attribute

materialization = None

meta class-attribute instance-attribute

meta = None

build_ddl

build_ddl(if_not_exists=True, or_replace=False)

Build a CREATE TABLE DDL statement directly from this model's properties.

Unlike :meth:get_ddl, this does not require root_model — it uses schema, fqn, table_type, table_properties, cluster_by, partition_cols, path, and comment directly from the dataclass fields.

Parameters:

Name	Type	Description	Default
if_not_exists	bool	Emit IF NOT EXISTS in the statement.	True
or_replace	bool	Emit OR REPLACE, replacing an existing table in place.	False

Returns:

Type	Description
str | None	DDL string, or None when schema is not set.

Raises:

Type	Description
ValueError	If or_replace is combined with if_not_exists.

Source code in src/kelp/service/model_manager.py

def build_ddl(self, if_not_exists: bool = True, or_replace: bool = False) -> str | None:
    """Build a CREATE TABLE DDL statement directly from this model's properties.

    Unlike :meth:`get_ddl`, this does not require ``root_model`` — it uses
    ``schema``, ``fqn``, ``table_type``, ``table_properties``,
    ``cluster_by``, ``partition_cols``, ``path``, and ``comment``
    directly from the dataclass fields.

    Args:
        if_not_exists: Emit ``IF NOT EXISTS`` in the statement.
        or_replace: Emit ``OR REPLACE``, replacing an existing table in place.

    Returns:
        DDL string, or ``None`` when ``schema`` is not set.

    Raises:
        ValueError: If ``or_replace`` is combined with ``if_not_exists``.
    """
    if or_replace and if_not_exists:
        raise ValueError("'or_replace' cannot be combined with 'if_not_exists'.")
    if not self.schema:
        return None

    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    target = self.fqn or self.name

    ddl = "CREATE "
    if or_replace:
        ddl += "OR REPLACE "
    ddl += f"{mapped_type} "
    if if_not_exists:
        ddl += "IF NOT EXISTS "
    ddl += f"{target} (\n{self.schema}\n)"

    if self.comment:
        ddl += f"\nCOMMENT '{self.comment}'"
    if self.path:
        ddl += f"\nLOCATION '{self.path}'"
    if self.cluster_by_auto:
        ddl += "\nCLUSTER BY (AUTO)"
    elif self.cluster_by:
        ddl += f"\nCLUSTER BY ({', '.join(self.cluster_by)})"
    elif self.partition_cols:
        ddl += f"\nPARTITIONED BY ({', '.join(self.partition_cols)})"
    if self.table_properties:
        props = ", ".join(f"'{k}'='{v}'" for k, v in self.table_properties.items())
        ddl += f"\nTBLPROPERTIES ({props})"

    return ddl

get_ddl

get_ddl(if_not_exists=True)

Source code in src/kelp/service/model_manager.py

def get_ddl(self, if_not_exists: bool = True) -> str | None:
    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    if self.root_model:
        return ModelManager.get_spark_schema_ddl(
            self.root_model,
            table_type=mapped_type,
            if_not_exists=if_not_exists,
        )
    # Fallback to building DDL from dataclass fields when root_model is not available
    return self.build_ddl(if_not_exists=if_not_exists)

get_replace_ddl

get_replace_ddl()

Build a CREATE OR REPLACE TABLE statement for this model.

Used by the replace full-refresh strategy: unlike DROP + CREATE, replacing keeps the table's identity, so Unity Catalog grants, tags and history survive the refresh.

Returns:

Type	Description
str | None	DDL string, or None when no schema is available.

Source code in src/kelp/service/model_manager.py

def get_replace_ddl(self) -> str | None:
    """Build a ``CREATE OR REPLACE TABLE`` statement for this model.

    Used by the ``replace`` full-refresh strategy: unlike ``DROP`` + ``CREATE``,
    replacing keeps the table's identity, so Unity Catalog grants, tags and
    history survive the refresh.

    Returns:
        DDL string, or ``None`` when no schema is available.
    """
    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    if self.root_model:
        return ModelManager.get_spark_schema_ddl(
            self.root_model,
            table_type=mapped_type,
            or_replace=True,
        )
    return self.build_ddl(if_not_exists=False, or_replace=True)

kelp.service.model_manager.KelpSdpModel dataclass

KelpSdpModel(
    name,
    table_type=None,
    comment=None,
    table_properties=None,
    spark_conf=None,
    path=None,
    partition_cols=None,
    cluster_by_auto=None,
    cluster_by=None,
    row_filter=None,
    auto_ttl=None,
    fqn=None,
    schema=None,
    schema_lite=None,
    dqx_quality=None,
    validation_table=None,
    quarantine_table=None,
    target_table=None,
    root_model=None,
    materialization=None,
    meta=None,
    expect_all=None,
    expect_all_or_fail=None,
    expect_all_or_drop=None,
    expect_all_or_quarantine=None,
)

Bases: KelpModel

expect_all class-attribute instance-attribute

expect_all = None

expect_all_or_fail class-attribute instance-attribute

expect_all_or_fail = None

expect_all_or_drop class-attribute instance-attribute

expect_all_or_drop = None

expect_all_or_quarantine class-attribute instance-attribute

expect_all_or_quarantine = None

name instance-attribute

name

table_type class-attribute instance-attribute

table_type = None

comment class-attribute instance-attribute

comment = None

table_properties class-attribute instance-attribute

table_properties = None

spark_conf class-attribute instance-attribute

spark_conf = None

path class-attribute instance-attribute

path = None

partition_cols class-attribute instance-attribute

partition_cols = None

cluster_by_auto class-attribute instance-attribute

cluster_by_auto = None

cluster_by class-attribute instance-attribute

cluster_by = None

row_filter class-attribute instance-attribute

row_filter = None

auto_ttl class-attribute instance-attribute

auto_ttl = None

fqn class-attribute instance-attribute

fqn = None

schema class-attribute instance-attribute

schema = None

schema_lite class-attribute instance-attribute

schema_lite = None

dqx_quality class-attribute instance-attribute

dqx_quality = None

validation_table class-attribute instance-attribute

validation_table = None

quarantine_table class-attribute instance-attribute

quarantine_table = None

target_table class-attribute instance-attribute

target_table = None

root_model class-attribute instance-attribute

root_model = None

materialization class-attribute instance-attribute

materialization = None

meta class-attribute instance-attribute

meta = None

params

params(exclude=None)

Source code in src/kelp/service/model_manager.py

def params(self, exclude: list[str] | None = None) -> dict[str, str]:
    exclude = exclude or []
    default_exclude = [
        "expect_all",
        "expect_all_or_drop",
        "expect_all_or_fail",
        "expect_all_or_quarantine",
    ]
    exclude = list(set(exclude) | set(default_exclude))
    return self.get_sdp_params(exclude=exclude)

params_raw

params_raw(exclude=None)

Source code in src/kelp/service/model_manager.py

def params_raw(self, exclude: list[str] | None = None) -> dict[str, str]:
    exclude = exclude or []
    return self.get_sdp_params(exclude=exclude)

params_cst

params_cst(exclude=None)

Source code in src/kelp/service/model_manager.py

def params_cst(self, exclude: list[str] | None = None) -> dict[str, str]:
    exclude = exclude or []
    default_exclude = ["expect_all_or_quarantine"]
    exclude = list(set(exclude) | set(default_exclude))
    return self.get_sdp_params(exclude=exclude)

get_sdp_params

get_sdp_params(exclude=None)

Source code in src/kelp/service/model_manager.py

def get_sdp_params(self, exclude: list[str] | None = None) -> dict[str, Any]:
    exclude = exclude or []
    params = {
        "name": self.fqn,
        "comment": self.comment,
        "spark_conf": self.spark_conf,
        "table_properties": self.table_properties,
        "path": self.path,
        "partition_cols": self.partition_cols,
        "cluster_by_auto": self.cluster_by_auto,
        "cluster_by": self.cluster_by,
        "schema": self.schema or None,
        "row_filter": self.row_filter,
        "auto_ttl": self.auto_ttl,
        "expect_all": self.expect_all,
        "expect_all_or_drop": self.expect_all_or_drop,
        "expect_all_or_fail": self.expect_all_or_fail,
        "expect_all_or_quarantine": self.expect_all_or_quarantine,
    }
    return {k: v for k, v in params.items() if (v is not None or "") and k not in exclude}

get_ddl

get_ddl(if_not_exists=False, or_refresh=True)

Source code in src/kelp/service/model_manager.py

def get_ddl(self, if_not_exists: bool = False, or_refresh: bool = True) -> str | None:
    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    return (
        ModelManager.get_spark_schema_ddl(
            self.root_model,
            table_type=mapped_type,
            if_not_exists=if_not_exists,
            or_refresh=or_refresh,
        )
        if self.root_model
        else None
    )

build_ddl

build_ddl(if_not_exists=True, or_replace=False)

Build a CREATE TABLE DDL statement directly from this model's properties.

Unlike :meth:get_ddl, this does not require root_model — it uses schema, fqn, table_type, table_properties, cluster_by, partition_cols, path, and comment directly from the dataclass fields.

Parameters:

Name	Type	Description	Default
if_not_exists	bool	Emit IF NOT EXISTS in the statement.	True
or_replace	bool	Emit OR REPLACE, replacing an existing table in place.	False

Returns:

Type	Description
str | None	DDL string, or None when schema is not set.

Raises:

Type	Description
ValueError	If or_replace is combined with if_not_exists.

Source code in src/kelp/service/model_manager.py

def build_ddl(self, if_not_exists: bool = True, or_replace: bool = False) -> str | None:
    """Build a CREATE TABLE DDL statement directly from this model's properties.

    Unlike :meth:`get_ddl`, this does not require ``root_model`` — it uses
    ``schema``, ``fqn``, ``table_type``, ``table_properties``,
    ``cluster_by``, ``partition_cols``, ``path``, and ``comment``
    directly from the dataclass fields.

    Args:
        if_not_exists: Emit ``IF NOT EXISTS`` in the statement.
        or_replace: Emit ``OR REPLACE``, replacing an existing table in place.

    Returns:
        DDL string, or ``None`` when ``schema`` is not set.

    Raises:
        ValueError: If ``or_replace`` is combined with ``if_not_exists``.
    """
    if or_replace and if_not_exists:
        raise ValueError("'or_replace' cannot be combined with 'if_not_exists'.")
    if not self.schema:
        return None

    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    target = self.fqn or self.name

    ddl = "CREATE "
    if or_replace:
        ddl += "OR REPLACE "
    ddl += f"{mapped_type} "
    if if_not_exists:
        ddl += "IF NOT EXISTS "
    ddl += f"{target} (\n{self.schema}\n)"

    if self.comment:
        ddl += f"\nCOMMENT '{self.comment}'"
    if self.path:
        ddl += f"\nLOCATION '{self.path}'"
    if self.cluster_by_auto:
        ddl += "\nCLUSTER BY (AUTO)"
    elif self.cluster_by:
        ddl += f"\nCLUSTER BY ({', '.join(self.cluster_by)})"
    elif self.partition_cols:
        ddl += f"\nPARTITIONED BY ({', '.join(self.partition_cols)})"
    if self.table_properties:
        props = ", ".join(f"'{k}'='{v}'" for k, v in self.table_properties.items())
        ddl += f"\nTBLPROPERTIES ({props})"

    return ddl

get_replace_ddl

get_replace_ddl()

Build a CREATE OR REPLACE TABLE statement for this model.

Used by the replace full-refresh strategy: unlike DROP + CREATE, replacing keeps the table's identity, so Unity Catalog grants, tags and history survive the refresh.

Returns:

Type	Description
str | None	DDL string, or None when no schema is available.

Source code in src/kelp/service/model_manager.py

def get_replace_ddl(self) -> str | None:
    """Build a ``CREATE OR REPLACE TABLE`` statement for this model.

    Used by the ``replace`` full-refresh strategy: unlike ``DROP`` + ``CREATE``,
    replacing keeps the table's identity, so Unity Catalog grants, tags and
    history survive the refresh.

    Returns:
        DDL string, or ``None`` when no schema is available.
    """
    mapped_type = _UC_TYPE.get(self.table_type.lower(), "TABLE") if self.table_type else "TABLE"
    if self.root_model:
        return ModelManager.get_spark_schema_ddl(
            self.root_model,
            table_type=mapped_type,
            or_replace=True,
        )
    return self.build_ddl(if_not_exists=False, or_replace=True)