fastflowtransform

FastFlowTransform package entry point.

Expose the package version and a few commonly-used types/APIs for convenience. Importing here is intentionally lightweight to avoid circular imports with CLI code.

EnvCtx dataclass

Stable environment context used for fingerprinting. Include only inputs that should invalidate compiled artifacts when they change.

Source code in src/fastflowtransform/fingerprint.py

@dataclass(frozen=True)
class EnvCtx:
    """
    Stable environment context used for fingerprinting.
    Include only inputs that should invalidate compiled artifacts when they change.
    """

    engine: str
    profile: str
    env_vars: Mapping[str, str]
    sources_json: str

    def to_payload(self) -> Mapping[str, Any]:
        return {
            "engine": self.engine,
            "profile": self.profile,
            "env": {k: self.env_vars.get(k, "") for k in sorted(self.env_vars.keys())},
            "sources": self.sources_json,
        }

relation_for

relation_for(node_name)

Map a logical node name to the physical relation (table/view name). Convention: - if the name ends with '.ff' → strip the suffix (e.g. 'users.ff' → 'users') - otherwise: return unchanged

Source code in src/fastflowtransform/core.py

def relation_for(node_name: str) -> str:
    """
    Map a logical node name to the physical relation (table/view name).
    Convention:
      - if the name ends with '.ff' → strip the suffix (e.g. 'users.ff' → 'users')
      - otherwise: return unchanged
    """
    return node_name[:-3] if node_name.endswith(".ff") else node_name

levels

levels(nodes)

Returns a level-wise topological ordering. - Each inner list contains nodes with no prerequisites inside the remaining graph (i.e. eligible to run in parallel). - Ordering within a level is lexicographically stable. - Validation for missing deps/cycles matches topo_sort.

Source code in src/fastflowtransform/dag.py

def levels(nodes: dict[str, Node]) -> list[list[str]]:
    """
    Returns a level-wise topological ordering.
    - Each inner list contains nodes with no prerequisites inside the remaining
      graph (i.e. eligible to run in parallel).
    - Ordering within a level is lexicographically stable.
    - Validation for missing deps/cycles matches topo_sort.
    """
    # Fehlende Deps einsammeln (nur Modell-Refs; sources sind keine Nodes)
    missing = {
        n.name: sorted({d for d in (n.deps or []) if d not in nodes})
        for n in nodes.values()
        if any(d not in nodes for d in (n.deps or []))
    }
    if missing:
        raise DependencyNotFoundError(missing)

    indeg = {k: 0 for k in nodes}
    out: dict[str, set[str]] = defaultdict(set)
    for n in nodes.values():
        for d in set(n.deps or []):
            out[d].add(n.name)
            indeg[n.name] += 1

    # Start-Level: alle 0-Indegree
    current = sorted([k for k, deg in indeg.items() if deg == 0])
    lvls: list[list[str]] = []
    seen_count = 0

    while current:
        lvls.append(current)
        next_zero: set[str] = set()
        for u in current:
            seen_count += 1
            for v in sorted(out.get(u, ())):
                indeg[v] -= 1
                if indeg[v] == 0:
                    next_zero.add(v)
        current = sorted(next_zero)

    if seen_count != len(nodes):
        cyclic = [k for k, deg in indeg.items() if deg > 0]
        raise ModelCycleError(f"Cycle detected among nodes: {', '.join(sorted(cyclic))}")
    return lvls

dq_test

dq_test(name=None, *, overwrite=False, params_model=None)

Decorator to register a custom data-quality test runner.

Usage:

from fastflowtransform import dq_test

@dq_test("email_domain_allowed")
def email_domain_allowed(executor, table, column, params):
    ...
    return True, None, "select ..."

If name is omitted, the function name is used:

@dq_test()
def email_sanity(executor, table, column, params):
    ...

# In project.yml / schema.yml: type: email_sanity

Params model:

class EmailTestParams(DQParamsBase):
    allowed_domains: list[str]

@dq_test("email_domain_allowed", params_model=EmailTestParams)
def email_domain_allowed(executor, table, column, params: EmailTestParams):
    ...

Parameters:

Name	Type	Description	Default
name	str | None	Optional explicit test name. If None, fn.name is used.	None
overwrite	bool	If True, allow overriding an existing test name.	False
params_model	type[BaseModel] | None	Optional Pydantic model to validate params. If omitted, DQParamsBase (extra='forbid') is used.	None

Source code in src/fastflowtransform/decorators.py

def dq_test(
    name: str | None = None,
    *,
    overwrite: bool = False,
    params_model: type[BaseModel] | None = None,
) -> Callable[[Callable[..., Any]], Runner]:
    """
    Decorator to register a custom data-quality test runner.

    Usage:

        from fastflowtransform import dq_test

        @dq_test("email_domain_allowed")
        def email_domain_allowed(executor, table, column, params):
            ...
            return True, None, "select ..."

    If `name` is omitted, the function name is used:

        @dq_test()
        def email_sanity(executor, table, column, params):
            ...

        # In project.yml / schema.yml: type: email_sanity

    Params model:

        class EmailTestParams(DQParamsBase):
            allowed_domains: list[str]

        @dq_test("email_domain_allowed", params_model=EmailTestParams)
        def email_domain_allowed(executor, table, column, params: EmailTestParams):
            ...

    Args:
        name: Optional explicit test name. If None, fn.__name__ is used.
        overwrite: If True, allow overriding an existing test name.
        params_model: Optional Pydantic model to validate `params`.
                      If omitted, DQParamsBase (extra='forbid') is used.
    """

    def decorator(fn: Callable[..., Any]) -> Runner:
        # Prefer attribute __name__ when available; fallback is just a placeholder.
        if name is not None:
            reg_name: str = name
        else:
            reg_name = cast(str, getattr(fn, "__name__", "<anonymous_test>"))

        pm = params_model or DQParamsBase

        # Central registration so test_cmd can pick up the params schema
        register_python_test(reg_name, fn, params_model=pm, overwrite=overwrite)

        # Attach a bit of metadata (not required, but can be handy for debugging/introspection)
        fn_any = cast(Any, fn)
        fn_any.__ff_test_name__ = reg_name
        fn_any.__ff_test_params_model__ = pm

        # Type-wise, fn already matches Runner's call signature at runtime
        return cast(Runner, fn)

    return decorator

engine_model

engine_model(*, only=None, env_match=None, **model_kwargs)

Env-aware decorator to register a Python model only when the current environment matches.

Parameters:

Name	Type	Description	Default
only	str | Iterable[str] | None	Backwards compatible engine filter based on FF_ENGINE (e.g. only="bigquery" or only=("duckdb", "postgres")).	None
env_match	Mapping[str, str] | None	Arbitrary environment match, e.g.: env_match={"FF_ENGINE": "bigquery", "FF_ENGINE_VARIANT": "bigframes"}	None

Source code in src/fastflowtransform/decorators.py

def engine_model(
    *,
    only: str | Iterable[str] | None = None,
    env_match: Mapping[str, str] | None = None,
    **model_kwargs: Any,
) -> Callable[[Callable[P, R_co]], HasFFMeta[P, R_co] | Callable[P, R_co]]:
    """
    Env-aware decorator to register a Python model only when the current
    environment matches.

    Args:
        only:
            Backwards compatible engine filter based on FF_ENGINE
            (e.g. only="bigquery" or only=("duckdb", "postgres")).
        env_match:
            Arbitrary environment match, e.g.:
                env_match={"FF_ENGINE": "bigquery", "FF_ENGINE_VARIANT": "bigframes"}
    """

    # Normalize "only" → allowed engine names (lowercased)
    allowed_engines: set[str] | None = None
    if only is not None:
        if isinstance(only, str):
            allowed_engines = {only.lower()}
        else:
            allowed_engines = {str(e).lower() for e in only}

    def should_register() -> bool:
        # 1) Check env_match if provided
        if env_match:
            for key, expected in env_match.items():
                if os.getenv(key) != expected:
                    return False

        # 2) Check FF_ENGINE against "only" if provided
        if allowed_engines is not None:
            current = os.getenv("FF_ENGINE", "").lower()
            if current not in allowed_engines:
                return False

        return True

    def deco(fn: Callable[P, R_co]) -> HasFFMeta[P, R_co] | Callable[P, R_co]:
        if should_register():
            # Register in REGISTRY and attach __ff_* metadata
            return model(**model_kwargs)(fn)
        # No registration in this env → return the plain function
        return fn

    return deco

model

model(name=None, deps=None, require=None, requires=None, *, tags=None, kind='python', materialized=None, meta=None)

Decorator to register a Python model.

Parameters:

Name	Type	Description	Default
name	str | None	Logical node name in the DAG (defaults to function name).	None
deps	Sequence[str] | None	Upstream node names (e.g., ['users.ff']).	None
require	Any | None	Required columns per dependency; accepted shapes mirror requires.	None
requires	Any | None	Alias for require (only one of require/requires may be set). - Single dependency: Iterable[str] of required columns from that dependency. - Multiple dependencies: Mapping[dep_name, Iterable[str]] (dep_name = logical name or physical relation).	None
tags	Sequence[str] | None	Optional tags for selection (e.g. ['demo','env']).	None
kind	str	Logical kind; defaults to 'python' (useful for selectors kind:python).	'python'
materialized	str | None	Shorthand for meta['materialized']; mirrors config(materialized='...').	None
meta	Mapping[str, Any] | None	Arbitrary metadata for executors/docs (merged with materialized if provided).	None

Source code in src/fastflowtransform/decorators.py

def model(
    name: str | None = None,
    deps: Sequence[str] | None = None,
    require: Any | None = None,
    requires: Any | None = None,
    *,
    tags: Sequence[str] | None = None,
    kind: str = "python",
    materialized: str | None = None,
    meta: Mapping[str, Any] | None = None,
) -> Callable[[Callable[P, R_co]], HasFFMeta[P, R_co]]:
    """
    Decorator to register a Python model.

    Args:
        name: Logical node name in the DAG (defaults to function name).
        deps: Upstream node names (e.g., ['users.ff']).
        require: Required columns per dependency; accepted shapes mirror `requires`.
        requires: Alias for `require` (only one of require/requires may be set).
            - Single dependency: Iterable[str] of required columns from that dependency.
            - Multiple dependencies: Mapping[dep_name, Iterable[str]]
              (dep_name = logical name or physical relation).
        tags: Optional tags for selection (e.g. ['demo','env']).
        kind: Logical kind; defaults to 'python' (useful for selectors kind:python).
        materialized: Shorthand for meta['materialized']; mirrors config(materialized='...').
        meta: Arbitrary metadata for executors/docs (merged with materialized if provided).
    """
    # Normalize the alias: allow only one of require/requires
    if require is not None and requires is not None:
        raise TypeError("Pass at most one of 'require' or 'requires', not both")

    effective_require = require if require is not None else requires

    def deco(func: Callable[P, R_co]) -> HasFFMeta[P, R_co]:
        f_any = cast(Any, func)

        fname = name or f_any.__name__
        fdeps = list(deps) if deps is not None else []

        # Attach metadata to the function (keeps backward compatibility)
        f_any.__ff_name__ = fname
        f_any.__ff_deps__ = fdeps

        # Normalize require and mirror it on the function and inside the registry
        req_norm = _normalize_require(fdeps, effective_require)
        f_any.__ff_require__ = req_norm  # useful for tooling/loaders
        REGISTRY.py_requires[fname] = req_norm  # executors read this directly

        f_any.__ff_tags__ = list(tags) if tags else []
        f_any.__ff_kind__ = kind or "python"

        metadata = dict(meta) if meta else {}
        if materialized is not None:
            metadata["materialized"] = materialized
        f_any.__ff_meta__ = metadata

        # Determine the source path (better error message if it fails)
        src: str | None = inspect.getsourcefile(func)
        if src is None:
            try:
                src = inspect.getfile(func)
            except Exception as e:
                raise ModuleLoadError(
                    f"Cannot determine source path for model '{fname}': {e}"
                ) from e

        f_any.__ff_path__ = Path(src).resolve()

        # Register the function
        REGISTRY.py_funcs[fname] = func
        return cast(HasFFMeta[P, R_co], func)

    return deco

build_env_ctx

build_env_ctx(*, engine, profile_name, relevant_env_keys=(), sources=None)

Construct an EnvCtx from engine/profile + a curated set of environment variables and the (normalized) sources.yml mapping. Only the provided environment keys are captured; all others are ignored.

Source code in src/fastflowtransform/fingerprint.py

def build_env_ctx(
    *,
    engine: str,
    profile_name: str,
    relevant_env_keys: Iterable[str] = (),
    sources: Mapping[str, Any] | None = None,
) -> EnvCtx:
    """
    Construct an EnvCtx from engine/profile + a curated set of environment variables
    and the (normalized) sources.yml mapping.
    Only the provided environment keys are captured; all others are ignored.
    """
    env_subset: dict[str, str] = {}
    for key in sorted(set(relevant_env_keys)):
        val = os.getenv(key)
        if val is not None:
            env_subset[key] = val
    return EnvCtx(
        engine=str(engine),
        profile=str(profile_name),
        env_vars=env_subset,
        sources_json=normalized_sources_blob(sources),
    )

fingerprint_py

fingerprint_py(*, node, func_src, env_ctx, dep_fps=None)

Compute a stable fingerprint for a Python model. Inputs: - node : Node or node name - func_src : normalized function source (use get_function_source) - env_ctx : EnvCtx or compatible mapping - dep_fps : mapping of dependency name → fingerprint

Source code in src/fastflowtransform/fingerprint.py

def fingerprint_py(
    *,
    node: Node | str,
    func_src: str,
    env_ctx: EnvCtx | Mapping[str, Any],
    dep_fps: Mapping[str, str] | None = None,
) -> str:
    """
    Compute a stable fingerprint for a Python model.
    Inputs:
      - node     : Node or node name
      - func_src : normalized function source (use get_function_source)
      - env_ctx  : EnvCtx or compatible mapping
      - dep_fps  : mapping of dependency name → fingerprint
    """
    n_name = node.name if isinstance(node, Node) else str(node)
    payload = {
        "kind": "python",
        "node": n_name,
        "relation": relation_for(n_name),
        "func_src": func_src.replace("\r\n", "\n").replace("\r", "\n").strip(),
        "env": env_ctx.to_payload() if isinstance(env_ctx, EnvCtx) else _as_primitive(env_ctx),
        "deps": _as_primitive(sorted((dep_fps or {}).items(), key=lambda kv: kv[0])),
    }
    return _hash_hex(_stable_dumps(payload))

fingerprint_sql

fingerprint_sql(*, node, rendered_sql, env_ctx, dep_fps=None)

Compute a stable fingerprint for a SQL model. Inputs: - node : Node or node name for stable identity and relation - rendered_sql : final SQL after templating (ref()/source() resolved as in executor) - env_ctx : EnvCtx or compatible mapping (engine, profile, selected env vars, sources) - dep_fps : mapping of dependency name → fingerprint (to invalidate downstream)

Source code in src/fastflowtransform/fingerprint.py

def fingerprint_sql(
    *,
    node: Node | str,
    rendered_sql: str,
    env_ctx: EnvCtx | Mapping[str, Any],
    dep_fps: Mapping[str, str] | None = None,
) -> str:
    """
    Compute a stable fingerprint for a SQL model.
    Inputs:
      - node         : Node or node name for stable identity and relation
      - rendered_sql : final SQL after templating (ref()/source() resolved as in executor)
      - env_ctx      : EnvCtx or compatible mapping (engine, profile, selected env vars, sources)
      - dep_fps      : mapping of dependency name → fingerprint (to invalidate downstream)
    """
    n_name = node.name if isinstance(node, Node) else str(node)
    payload = {
        "kind": "sql",
        "node": n_name,
        "relation": relation_for(n_name),
        "sql": _normalize_sql(rendered_sql),
        "env": env_ctx.to_payload() if isinstance(env_ctx, EnvCtx) else _as_primitive(env_ctx),
        "deps": _as_primitive(sorted((dep_fps or {}).items(), key=lambda kv: kv[0])),
    }
    return _hash_hex(_stable_dumps(payload))

get_function_source

get_function_source(func)

Return a best-effort, stable source string for a Python callable.

Strategy (in order): 1) inspect.getsource(func) → dedented string 2) Read the defining file (co_filename) and slice starting at co_firstlineno until the next top-level def/class (heuristic). Dedent as needed. 3) Final fallback: combine qualified name and bytecode to ensure stability.

This ensures fingerprinting works even for dynamically loaded modules, lambdas, or environments where inspect cannot read the original file (e.g., zipimport).

Source code in src/fastflowtransform/fingerprint.py

def get_function_source(func: Any) -> str:
    """
    Return a best-effort, stable source string for a Python callable.

    Strategy (in order):
    1) inspect.getsource(func)  → dedented string
    2) Read the defining file (co_filename) and slice starting at co_firstlineno
       until the next top-level def/class (heuristic). Dedent as needed.
    3) Final fallback: combine qualified name and bytecode to ensure stability.

    This ensures fingerprinting works even for dynamically loaded modules, lambdas,
    or environments where inspect cannot read the original file (e.g., zipimport).
    """
    # 1) The happy path
    try:
        src = inspect.getsource(func)
        return textwrap.dedent(src).strip()
    except Exception:
        pass

    # 2) Slice from file using code object hints
    try:
        code = getattr(func, "__code__", None)
        if code and isinstance(code.co_firstlineno, int) and code.co_filename:
            file_path = Path(code.co_filename)
            # Read as binary + decode to be robust to odd encodings
            with open(file_path, "rb") as fh:
                raw = fh.read()
            text = raw.decode("utf-8", errors="replace")
            start = max(code.co_firstlineno - 1, 0)

            lines = text.splitlines()
            # Heuristic: collect until the next top-level def/class (same or less indent)
            buf: list[str] = []
            base_indent = None
            for idx in range(start, len(lines)):
                line = lines[idx]
                buf.append(line)
                # capture base indentation from the first non-empty line
                if base_indent is None and line.strip():
                    base_indent = len(line) - len(line.lstrip())
                # stop when we hit a new top-level def/class after the first line
                if (
                    idx > start
                    and line
                    and not line.startswith(" " * (base_indent or 0))
                    and line.lstrip().startswith(("def ", "class ", "@"))
                ):
                    buf.pop()  # don't include the new top-level symbol
                    break
            sliced = "\n".join(buf)
            return textwrap.dedent(sliced).strip()
    except Exception:
        pass

    # 3) Last resort: qualname + bytecode hash
    try:
        qual = getattr(func, "__qualname__", getattr(func, "__name__", "anonymous"))
        bc = getattr(getattr(func, "__code__", None), "co_code", b"")
        payload = f"{qual}\nBYTECODE:{hashlib.sha256(bc).hexdigest()}"
        return payload
    except Exception:
        return "UNKNOWN_FUNCTION"

normalized_sources_blob

normalized_sources_blob(sources)

Return a stable JSON blob for a sources.yml mapping. Keys are sorted recursively; absent input becomes "{}".

Source code in src/fastflowtransform/fingerprint.py

def normalized_sources_blob(sources: Mapping[str, Any] | None) -> str:
    """
    Return a stable JSON blob for a sources.yml mapping.
    Keys are sorted recursively; absent input becomes "{}".
    """
    return _stable_dumps(sources or {})