fastflowtransform.contracts.runtime.snowflake_snowpark

SnowflakeSnowparkRuntimeContracts

Bases: BaseRuntimeContracts

Runtime schema contracts for Snowflake (Snowpark).

SQL models

• off: CREATE OR REPLACE TABLE AS
• verify: CTAS then compare information_schema.columns
• cast: CTAS via subquery with CAST(...) projections, then verify

Python models (Snowpark DataFrame): - verify: write.save_as_table(... overwrite) then verify - cast: create temp view from DF, do CTAS with casts, then verify

Source code in src/fastflowtransform/contracts/runtime/snowflake_snowpark.py

class SnowflakeSnowparkRuntimeContracts(BaseRuntimeContracts):
    """
    Runtime schema contracts for Snowflake (Snowpark).

    SQL models:
      - off:   CREATE OR REPLACE TABLE AS <select_body>
      - verify: CTAS then compare information_schema.columns
      - cast:  CTAS via subquery with CAST(...) projections, then verify

    Python models (Snowpark DataFrame):
      - verify: write.save_as_table(... overwrite) then verify
      - cast:   create temp view from DF, do CTAS with casts, then verify
    """

    def __init__(self, executor: ContractExecutor):
        super().__init__(executor)

    # --- tiny helpers -----------------------------------------------------

    def _exec(self, sql: str) -> None:
        """
        Snowpark executes on action; collect() is the simplest action.
        """
        res = self.executor._execute_sql(sql)
        collect = getattr(res, "collect", None)
        if callable(collect):
            collect()
        else:
            # best-effort fallback (shouldn't happen for real Snowpark)
            _ = res

    def _safe_tmp_name(self, base: str) -> str:
        txt = re.sub(r"[^A-Za-z0-9_]+", "_", base or "tmp").strip("_")
        if not txt:
            txt = "tmp"
        return f"__FF_CONTRACT_{txt}".upper()

    def _verify(
        self,
        *,
        table: str,
        expected: Mapping[str, str],
        cfg: RuntimeContractConfig,
    ) -> None:
        if not expected:
            return

        actual = self.executor.introspect_table_physical_schema(table)
        exp_lower = {k.lower(): v for k, v in expected.items()}

        problems: list[str] = []

        for col, expected_type in expected.items():
            key = col.lower()
            if key not in actual:
                problems.append(f"- missing column {col!r}")
                continue
            got = actual[key]
            if not _types_match(expected_type, got):
                problems.append(f"- column {col!r}: expected {expected_type!r}, got {got!r}")

        if not cfg.allow_extra_columns:
            extras = [c for c in actual if c not in exp_lower]
            if extras:
                problems.append(f"- extra columns present: {sorted(extras)}")

        if problems:
            raise RuntimeError(
                f"[contracts] Snowflake schema enforcement failed for {table}:\n"
                + "\n".join(problems)
            )

    def _ctas_raw(self, target: str, select_body: str) -> None:
        self._exec(f"CREATE OR REPLACE TABLE {target} AS {select_body}")

    def _ctas_cast_via_subquery(
        self,
        *,
        ctx: RuntimeContractContext,
        select_body: str,
        expected: Mapping[str, str],
    ) -> None:
        if not expected:
            self._ctas_raw(ctx.physical_table, select_body)
            return

        exp_lower = {k.lower(): v for k, v in expected.items()}
        projections: list[str] = [f"CAST({col} AS {typ}) AS {col}" for col, typ in expected.items()]

        if ctx.config.allow_extra_columns:
            tmp = self._safe_tmp_name(f"{ctx.relation}_tmp")
            # temp table holds the raw output so we can discover extras
            self._exec(f"CREATE OR REPLACE TEMPORARY TABLE {tmp} AS {select_body}")

            try:
                actual = self.executor.introspect_table_physical_schema(tmp)
                for c in actual:
                    if c not in exp_lower:
                        # c is already lowercased key; emit identifier as-is
                        projections.append(c)

                proj_sql = ", ".join(projections)
                self._exec(
                    f"CREATE OR REPLACE TABLE {ctx.physical_table} AS SELECT {proj_sql} FROM {tmp}"
                )
            finally:
                with suppress(Exception):
                    self._exec(f"DROP TABLE IF EXISTS {tmp}")
        else:
            proj_sql = ", ".join(projections)
            wrapped = f"SELECT {proj_sql} FROM ({select_body}) AS SRC"
            self._ctas_raw(ctx.physical_table, wrapped)

    # --- BaseRuntimeContracts hooks ---------------------------------------

    def apply_sql_contracts(
        self,
        *,
        ctx: RuntimeContractContext,
        select_body: str,
    ) -> None:
        expected = expected_physical_schema(executor=self.executor, contract=ctx.contract)
        mode = ctx.config.mode

        if mode == "off" or not expected:
            self._ctas_raw(ctx.physical_table, select_body)
            return

        if mode == "cast":
            if not expected:
                raise RuntimeError(
                    f"[contracts] cast mode enabled for {ctx.relation!r} "
                    "but no physical schema could be resolved."
                )
            self._ctas_cast_via_subquery(ctx=ctx, select_body=select_body, expected=expected)
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
            return

        if mode == "verify":
            self._ctas_raw(ctx.physical_table, select_body)
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
            return

        self._ctas_raw(ctx.physical_table, select_body)

    def materialize_python(
        self,
        *,
        ctx: RuntimeContractContext,
        df: Any,
    ) -> bool:
        """
        Snowpark Python models return a Snowpark DataFrame.

        - verify: write to table via save_as_table then verify
        - cast:   create temp view from df, then CTAS with CASTs, then verify

        Return True because we fully materialized the target table here.
        """
        mode = ctx.config.mode
        if mode == "off":
            return False

        expected = expected_physical_schema(executor=self.executor, contract=ctx.contract)

        if mode == "cast" and not expected:
            raise RuntimeError(
                f"[contracts] cast mode enabled for {ctx.relation!r} "
                "but no physical schema could be resolved."
            )

        # Basic Snowpark DF surface check
        if not (
            hasattr(df, "write")
            and hasattr(df, "schema")
            and callable(getattr(df, "collect", None))
        ):
            return False

        if mode == "verify":
            # Let Snowpark write in its native path, then verify
            df.write.save_as_table(ctx.physical_table, mode="overwrite")
            if expected:
                self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
            return True

        if mode == "cast":
            tmp_view = self._safe_tmp_name(f"{ctx.relation}_src")
            create_view = getattr(df, "create_or_replace_temp_view", None)
            if not callable(create_view):
                # fallback: if temp views aren't available, don't take over
                return False

            create_view(tmp_view)
            try:
                select_body = f"SELECT * FROM {tmp_view}"
                self._ctas_cast_via_subquery(ctx=ctx, select_body=select_body, expected=expected)
                self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
                return True
            finally:
                with suppress(Exception):
                    self._exec(f"DROP VIEW IF EXISTS {tmp_view}")

        return False

    def verify_after_materialization(self, *, ctx: RuntimeContractContext) -> None:
        expected = expected_physical_schema(executor=self.executor, contract=ctx.contract)
        if not expected:
            return
        if ctx.config.mode not in {"verify", "cast"}:
            return
        self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)

materialize_python

materialize_python(*, ctx, df)

Snowpark Python models return a Snowpark DataFrame.

• verify: write to table via save_as_table then verify
• cast: create temp view from df, then CTAS with CASTs, then verify

Return True because we fully materialized the target table here.

Source code in src/fastflowtransform/contracts/runtime/snowflake_snowpark.py

def materialize_python(
    self,
    *,
    ctx: RuntimeContractContext,
    df: Any,
) -> bool:
    """
    Snowpark Python models return a Snowpark DataFrame.

    - verify: write to table via save_as_table then verify
    - cast:   create temp view from df, then CTAS with CASTs, then verify

    Return True because we fully materialized the target table here.
    """
    mode = ctx.config.mode
    if mode == "off":
        return False

    expected = expected_physical_schema(executor=self.executor, contract=ctx.contract)

    if mode == "cast" and not expected:
        raise RuntimeError(
            f"[contracts] cast mode enabled for {ctx.relation!r} "
            "but no physical schema could be resolved."
        )

    # Basic Snowpark DF surface check
    if not (
        hasattr(df, "write")
        and hasattr(df, "schema")
        and callable(getattr(df, "collect", None))
    ):
        return False

    if mode == "verify":
        # Let Snowpark write in its native path, then verify
        df.write.save_as_table(ctx.physical_table, mode="overwrite")
        if expected:
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
        return True

    if mode == "cast":
        tmp_view = self._safe_tmp_name(f"{ctx.relation}_src")
        create_view = getattr(df, "create_or_replace_temp_view", None)
        if not callable(create_view):
            # fallback: if temp views aren't available, don't take over
            return False

        create_view(tmp_view)
        try:
            select_body = f"SELECT * FROM {tmp_view}"
            self._ctas_cast_via_subquery(ctx=ctx, select_body=select_body, expected=expected)
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
            return True
        finally:
            with suppress(Exception):
                self._exec(f"DROP VIEW IF EXISTS {tmp_view}")

    return False

build_context

build_context(*, node, relation, physical_table, contract, project_contracts, is_incremental=False)

Build a RuntimeContractContext with the correct RuntimeContractConfig.

The caller (run-engine) decides which contract applies and passes: - node: the fft Node being built - relation: logical name (typically node.name) - physical_table: fully-qualified identifier used in SQL - contract: per-table ContractsFileModel, or None - project_contracts: parsed project-level contracts.yml, or None

Source code in src/fastflowtransform/contracts/runtime/base.py

def build_context(
    self,
    *,
    node: Node,
    relation: str,
    physical_table: str,
    contract: ContractsFileModel | None,
    project_contracts: ProjectContractsModel | None,
    is_incremental: bool = False,
) -> RuntimeContractContext:
    """
    Build a RuntimeContractContext with the correct RuntimeContractConfig.

    The caller (run-engine) decides which contract applies and passes:
      - node:          the fft Node being built
      - relation:      logical name (typically node.name)
      - physical_table: fully-qualified identifier used in SQL
      - contract:      per-table ContractsFileModel, or None
      - project_contracts: parsed project-level contracts.yml, or None
    """
    # Use the contract's declared table name if present, otherwise fall
    # back to the logical relation name for project-level overrides.
    table_key = contract.table if contract is not None else relation

    cfg = resolve_runtime_contract_config(
        table_name=table_key,
        contract=contract,
        project_contracts=project_contracts,
    )

    return RuntimeContractContext(
        node=node,
        relation=relation,
        physical_table=physical_table,
        contract=contract,
        project_contracts=project_contracts,
        config=cfg,
        is_incremental=is_incremental,
    )

coerce_frame_schema

coerce_frame_schema(df, ctx)

Optional hook for Python models: given a DataFrame-like object and the RuntimeContractContext, return a new frame whose column types have been coerced to match the expected physical schema (where reasonable).

Default implementation is a no-op. Engine-specific subclasses may override this (e.g. DuckDB + pandas).

Source code in src/fastflowtransform/contracts/runtime/base.py

def coerce_frame_schema(self, df: Any, ctx: RuntimeContractContext) -> Any:
    """
    Optional hook for Python models: given a DataFrame-like object and the
    RuntimeContractContext, return a new frame whose column types have been
    coerced to match the expected physical schema (where reasonable).

    Default implementation is a no-op. Engine-specific subclasses may
    override this (e.g. DuckDB + pandas).
    """
    return df