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

model

model(name=None, deps=None, require=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	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,
    *,
    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:
            - 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).
    """

    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, 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 {})