Skip to content

Tables

kelp.tables

Generic model metadata API for use in any Spark job.

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 
31
32
33
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

Runner 

Runner()
Source code in src/kelp/tables/materialization/runner.py 
96
97
def __init__(self):
    self.runlog = RunLog()

runlog instance-attribute 

runlog = RunLog()

plan_one 

plan_one(name)
Source code in src/kelp/tables/materialization/runner.py 
 99
100
101
102
def plan_one(self, name: str) -> list[str]:
    if name not in _REGISTRY:
        raise KeyError(f"Unknown model: {name}")
    return [name]

plan_all 

plan_all()
Source code in src/kelp/tables/materialization/runner.py 
104
105
def plan_all(self) -> list[str]:
    return _toposort()

run_one 

run_one(name)
Source code in src/kelp/tables/materialization/runner.py 
107
108
109
110
def run_one(self, name: str):
    if name not in _REGISTRY:
        raise KeyError(f"Unknown model: {name}")
    return self._run_model(name)

run_all 

run_all(full_refresh=False)
Source code in src/kelp/tables/materialization/runner.py 
112
113
114
def run_all(self, full_refresh: bool = False) -> None:
    for name in self.plan_all():
        self._run_model(name, full_refresh=full_refresh)

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 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
54
55
56
57
58
59
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 
49
50
51
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 
62
63
64
65
66
67
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 
13
14
15
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 
18
19
20
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 
23
24
25
26
27
28
29
30
31
32
33
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 
36
37
38
39
40
41
42
43
44
45
46
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 
70
71
72
73
74
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 
77
78
79
80
81
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,
    full_refresh=False,
    sync_metadata=True,
    apply_vacuum=True,
    vacuum_lite=True,
    apply_optimize=True,
    apply_quality_checks=True,
    spark=None,
)

Materialize a DataFrame to Delta Lake based on materialization config.

Strategy: - Resolve model metadata by table name (if available). - Merge metadata materialization config with passed runtime config. - Dispatch to append/overwrite or merge materializers.

Parameters:

Name Type Description Default
dataframe DataFrame

DataFrame to materialize.

 required
name str

Kelp model name or fully qualified table name.

 required
config ModelMaterializationConfig | None

Optional runtime override materialization config.

 None
full_refresh bool

Whether to perform a full refresh, which may be prevented by model config.

 False
sync_metadata bool

Whether to perform metadata sync after materialization.

 True
apply_vacuum bool

Whether to apply VACUUM after materialization.

 True
vacuum_lite bool

Whether to use VACUUM LITE (if False, full VACUUM is applied). Only applicable if apply_vacuum is True.

 True
apply_optimize bool

Whether to apply OPTIMIZE after materialization.

 True
spark SparkSession | None

Optional SparkSession to use for materialization. If not provided, the active SparkSession will be used.

 None

Returns:

Type Description
DataFrame

The same input DataFrame (for chaining).
Source code in src/kelp/tables/materialization/factory.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def materialize(
    *,
    dataframe: DataFrame,
    name: str,
    config: ModelMaterializationConfig | None = None,
    full_refresh: bool = False,
    sync_metadata: bool = True,
    apply_vacuum: bool = True,
    vacuum_lite: bool = True,
    apply_optimize: bool = True,
    apply_quality_checks: bool = True,
    spark: SparkSession | None = None,
) -> DataFrame:
    """Materialize a DataFrame to Delta Lake based on materialization config.

    Strategy:
    - Resolve model metadata by table name (if available).
    - Merge metadata materialization config with passed runtime config.
    - Dispatch to append/overwrite or merge materializers.

    Args:
        dataframe: DataFrame to materialize.
        name: Kelp model name or fully qualified table name.
        config: Optional runtime override materialization config.
        full_refresh: Whether to perform a full refresh, which may be prevented by model config.
        sync_metadata: Whether to perform metadata sync after materialization.
        apply_vacuum: Whether to apply VACUUM after materialization.
        vacuum_lite: Whether to use VACUUM LITE (if False, full VACUUM is applied). Only applicable if apply_vacuum is True.
        apply_optimize: Whether to apply OPTIMIZE after materialization.
        spark: Optional SparkSession to use for materialization. If not provided, the active SparkSession will be used.

    Returns:
        The same input DataFrame (for chaining).
    """
    spark = spark or SparkSession.getActiveSession()
    if spark is None:
        raise RuntimeError("No active SparkSession available for materialization.")

    resolved = _resolve_materialization_inputs(table_name=name, config=config)

    logger.info(
        "Materializing '%s' to target '%s' with config: %s",
        name,
        resolved.target_name,
        resolved.effective_config.model_dump_json(),
    )

    result_df = dataframe
    if apply_quality_checks and resolved.dqx_quality and resolved.dqx_quality.checks:
        logger.debug(
            "Model '%s' has DQX checks defined. Applying quality checks before materialization.",
            resolved.target_name,
        )
        ensure_dqx_installed()

        from kelp.tables.quality_validation.dqx import apply_dqx_quality_checks

        dqx = resolved.dqx_quality
        quarantine_fqn = (
            resolved.kelp_model.quarantine_table
            if resolved.kelp_model and dqx.spark_quarantine
            else None
        )
        result_df = apply_dqx_quality_checks(
            dataframe=result_df,
            checks=dqx.checks,
            violation_action=dqx.spark_violation_action,
            target_table=resolved.target_name,
            quarantine_enabled=bool(dqx.spark_quarantine),
            quarantine_fqn=quarantine_fqn,
        )

    _materialize_model(
        spark=spark,
        dataframe=result_df,
        target_name=resolved.target_name,
        create_table_ddl=resolved.create_table_ddl,
        model_name=resolved.model_name,
        effective_config=resolved.effective_config,
        full_refresh=full_refresh,
    )

    # Apply metadata sync for the model after materialization
    if resolved.kelp_model and sync_metadata:
        # Skip if not on databricks
        from kelp.utils.databricks import on_databricks

        if on_databricks():
            _sync_metadata(spark, resolved.kelp_model)
        else:
            logger.info(
                "Skipping metadata sync for model '%s' since not running on Databricks",
                resolved.kelp_model.name,
            )

    _perform_maintenance(
        spark,
        resolved.target_name,
        apply_vacuum=apply_vacuum,
        vacuum_lite=vacuum_lite,
        apply_optimize=apply_optimize,
    )

    return dataframe

materialized 

materialized(
    func=None,
    *,
    name=None,
    config=None,
    depends_on=None,
    full_refresh=False,
    apply_vacuum=True,
    vacuum_lite=True,
    apply_optimize=True,
    apply_quality_checks=True,
)

Decorator that materializes the returned DataFrame.

Model matching uses name when provided; otherwise the wrapped function name is used.

Parameters:

Name Type Description Default
name str | None

Optional kelp model/table name.

 None
config ModelMaterializationConfig | dict | None

Optional materialization override config.

 None

Returns:

Type Description
Callable[[Callable[..., DataFrame]], Callable[..., DataFrame]]

Decorated callable returning the same DataFrame after materialization.
Source code in src/kelp/tables/materialization/decorator.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def materialized(
    func: Callable[..., DataFrame] | None = None,
    *,
    name: str | None = None,
    config: ModelMaterializationConfig | dict | None = None,
    depends_on: list[str] | None = None,
    full_refresh: bool = False,
    apply_vacuum: bool = True,
    vacuum_lite: bool = True,
    apply_optimize: bool = True,
    apply_quality_checks: bool = True,
) -> Callable[[Callable[..., DataFrame]], Callable[..., DataFrame]]:
    """Decorator that materializes the returned DataFrame.

    Model matching uses `name` when provided; otherwise the wrapped function
    name is used.

    Args:
        name: Optional kelp model/table name.
        config: Optional materialization override config.

    Returns:
        Decorated callable returning the same DataFrame after materialization.
    """
    cfg = ModelMaterializationConfig(**config) if isinstance(config, dict) else 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)

            spark = SparkSession.getActiveSession()
            if spark is None:
                raise RuntimeError("No active SparkSession available for materialization.")

            resolved_inputs = _resolve_materialization_inputs(
                table_name=target_name,
                config=cfg,
            )
            resolved_target_name = resolved_inputs.target_name

            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__}."
                )

            materialize(
                spark=spark,
                dataframe=result,
                name=target_name,
                config=cfg,
                full_refresh=runtime_full_refresh,
                apply_vacuum=apply_vacuum,
                vacuum_lite=vacuum_lite,
                apply_optimize=apply_optimize,
                apply_quality_checks=apply_quality_checks,
            )
            return result

        effective_name = target_name.split(".")[-1]
        _REGISTRY[effective_name] = ModelSpec(
            name=effective_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)

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

Returns:

Type Description
str | None

DDL string, or None when schema is not set.
Source code in src/kelp/service/model_manager.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def build_ddl(self, if_not_exists: bool = True) -> 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.

    Returns:
        DDL string, or ``None`` when ``schema`` is not set.
    """
    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 = f"CREATE {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 
90
91
92
93
94
95
96
97
98
99
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)

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 
109
110
111
112
113
114
115
116
117
118
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 
120
121
122
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 
124
125
126
127
128
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
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 
151
152
153
154
155
156
157
158
159
160
161
162
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)

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

Returns:

Type Description
str | None

DDL string, or None when schema is not set.
Source code in src/kelp/service/model_manager.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def build_ddl(self, if_not_exists: bool = True) -> 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.

    Returns:
        DDL string, or ``None`` when ``schema`` is not set.
    """
    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 = f"CREATE {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