fastflowtransform.contracts.runtime.postgres

PostgresRuntimeContracts

Bases: BaseRuntimeContracts

Runtime schema contracts for Postgres.

Like DuckRuntimeContracts, this composes an executor but can use Postgres-specific attributes (engine/schema) when doing Python DF staging.

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

class PostgresRuntimeContracts(BaseRuntimeContracts):
    """
    Runtime schema contracts for Postgres.

    Like DuckRuntimeContracts, this composes an executor but can use
    Postgres-specific attributes (engine/schema) when doing Python DF staging.
    """

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

    # --- helpers ---------------------------------------------------------

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

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

        problems: list[str] = []

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

            try:
                exp_canon = _pg_expected_canon(self.executor, expected_type)
            except ValueError as e:
                problems.append(f"- column {expected_col!r}: {e}")
                continue

            got_canon = actual[key]
            if got_canon != exp_canon:
                problems.append(
                    f"- column {expected_col!r}: expected type {expected_type!r} "
                    f"(canon={exp_canon!r}), got (canon={got_canon!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] Postgres schema enforcement failed for {table}:\n"
                + "\n".join(problems)
            )

    def _drop_create_ctas(self, target: str, select_body: str) -> None:
        # Postgres: no CREATE OR REPLACE TABLE
        self.executor._execute_sql(f"drop table if exists {target} cascade")
        self.executor._execute_sql(f"create table {target} as {select_body}")
        # best-effort stats
        with suppress(Exception):
            self.executor._execute_sql(f"analyze {target}")

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

        exp_lower = {k.lower(): v for k, v in expected.items()}

        # Always quote columns in projections
        projections: list[str] = [
            f"cast({_q_ident(col)} as {typ}) as {_q_ident(col)}" for col, typ in expected.items()
        ]

        if ctx.config.allow_extra_columns:
            # stage into a real table (not TEMP), because PostgresExecutor may use
            # different pooled connections per statement.
            tmp = _safe_tmp_name("__ff_contract_tmp_", ctx.relation)
            qtmp = _q_ident(tmp)

            self.executor._execute_sql(f"drop table if exists {qtmp} cascade")
            self.executor._execute_sql(f"create table {qtmp} as {select_body}")

            try:
                actual = self.executor.introspect_table_physical_schema(tmp)
                for c in actual:
                    if c not in exp_lower:
                        # actual keys are lowercase; we need the real column name as stored.
                        # safest: select by identifier using the lowercase key as-is.
                        projections.append(_q_ident(c))

                proj_sql = ", ".join(projections)
                self._drop_create_ctas(
                    ctx.physical_table,
                    f"select {proj_sql} from {qtmp}",
                )
            finally:
                with suppress(Exception):
                    self.executor._execute_sql(f"drop table if exists {qtmp} cascade")
        else:
            proj_sql = ", ".join(projections)
            wrapped = f"select {proj_sql} from ({select_body}) as src"
            self._drop_create_ctas(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._drop_create_ctas(ctx.physical_table, select_body)
            return

        if mode == "cast":
            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._drop_create_ctas(ctx.physical_table, select_body)
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)
            return

        # unknown mode -> behave like off
        self._drop_create_ctas(ctx.physical_table, select_body)

    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)

    def materialize_python(
        self,
        *,
        ctx: RuntimeContractContext,
        df: Any,
    ) -> bool:
        """
        Enforce contracts for pandas DataFrames by staging via to_sql, then
        creating the target via Postgres CASTs (cast mode) or verifying (verify mode).
        """
        mode = ctx.config.mode
        if mode == "off":
            return False
        if not isinstance(df, pd.DataFrame):
            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."
            )

        # We rely on PostgresExecutor providing `.engine` and `.schema` (like your executor does).
        engine = getattr(self.executor, "engine", None)
        schema = getattr(self.executor, "schema", None)
        if engine is None:
            return False

        tmp = _safe_tmp_name("__ff_py_src_", ctx.relation)
        qtmp = _q_ident(tmp)

        # 1) stage df -> table
        with suppress(Exception):
            self.executor._execute_sql(f"drop table if exists {qtmp} cascade")

        df.to_sql(
            tmp,
            engine,
            if_exists="replace",
            index=False,
            schema=schema,
            method="multi",
        )

        try:
            select_body = f"select * from {qtmp}"

            if mode == "cast":
                self._ctas_cast_via_subquery(
                    ctx=ctx,
                    select_body=select_body,
                    expected=expected,
                )
                self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)

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

            return True
        finally:
            with suppress(Exception):
                self.executor._execute_sql(f"drop table if exists {qtmp} cascade")

materialize_python

materialize_python(*, ctx, df)

Enforce contracts for pandas DataFrames by staging via to_sql, then creating the target via Postgres CASTs (cast mode) or verifying (verify mode).

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

def materialize_python(
    self,
    *,
    ctx: RuntimeContractContext,
    df: Any,
) -> bool:
    """
    Enforce contracts for pandas DataFrames by staging via to_sql, then
    creating the target via Postgres CASTs (cast mode) or verifying (verify mode).
    """
    mode = ctx.config.mode
    if mode == "off":
        return False
    if not isinstance(df, pd.DataFrame):
        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."
        )

    # We rely on PostgresExecutor providing `.engine` and `.schema` (like your executor does).
    engine = getattr(self.executor, "engine", None)
    schema = getattr(self.executor, "schema", None)
    if engine is None:
        return False

    tmp = _safe_tmp_name("__ff_py_src_", ctx.relation)
    qtmp = _q_ident(tmp)

    # 1) stage df -> table
    with suppress(Exception):
        self.executor._execute_sql(f"drop table if exists {qtmp} cascade")

    df.to_sql(
        tmp,
        engine,
        if_exists="replace",
        index=False,
        schema=schema,
        method="multi",
    )

    try:
        select_body = f"select * from {qtmp}"

        if mode == "cast":
            self._ctas_cast_via_subquery(
                ctx=ctx,
                select_body=select_body,
                expected=expected,
            )
            self._verify(table=ctx.physical_table, expected=expected, cfg=ctx.config)

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

        return True
    finally:
        with suppress(Exception):
            self.executor._execute_sql(f"drop table if exists {qtmp} cascade")

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