fastflowtransform.core

Registry

Source code in src/fastflowtransform/core.py

class Registry:
    def __init__(self):
        self.nodes: dict[str, Node] = {}
        self.py_funcs: dict[str, Callable] = {}
        self.project_dir: Path | None = None
        self.env = None
        self.sources: dict[str, dict[str, Any]] = {}
        self.py_requires: dict[str, dict[str, set[str]]] = {}
        self.macros: dict[str, Path] = {}  # macro_name -> file path
        self.project_vars: dict[str, Any] = {}  # project.yml: vars
        self.cli_vars: dict[str, Any] = {}  # CLI --vars overrides
        self.active_engine: str | None = None

    def get_project_dir(self) -> Path:
        """Return the project directory after load_project(), or raise if not set."""
        if self.project_dir is None:
            raise RuntimeError("Project directory not initialized. Call load_project() first.")
        return self.project_dir

    def get_env(self) -> Environment:
        """Return the initialized Jinja Environment, or raise if not loaded."""
        if self.env is None:
            raise RuntimeError("Jinja environment not initialized. Call load_project() first.")
        return self.env

    def get_node(self, name: str) -> Node:
        # exact match
        n = self.nodes.get(name)
        if n:
            return n
        # common aliases
        if name.endswith(".ff") and name in self.nodes:
            return self.nodes[name]
        alt = f"{name}.ff"
        n = self.nodes.get(alt)
        if n:
            return n
        raise KeyError(name)

    def set_cli_vars(self, overrides: dict[str, Any]) -> None:
        """Set CLI --vars overrides (highest precedence)."""
        self.cli_vars = dict(overrides or {})

    def set_active_engine(self, engine: str | None) -> None:
        """Store active engine hint (case-insensitive) for conditional loading."""
        self.active_engine = engine.lower().strip() if isinstance(engine, str) else None

    def _lookup_storage_meta(self, node_name: str) -> dict[str, Any]:
        """
        Return storage metadata for a given node (if configured in project.yml).
        Accepts names with or without trailing '.ff'.
        """
        return storage.get_model_storage(node_name)

    def _current_engine(self) -> str | None:
        """
        Determine the active engine in precedence order:
        1) Explicit hint via set_active_engine()
        2) Environment variable FF_ENGINE
        3) project.yml vars → engine
        4) CLI --vars {engine: ...}
        """
        if self.active_engine:
            return self.active_engine

        env_engine = os.getenv("FF_ENGINE")
        if isinstance(env_engine, str) and env_engine.strip():
            return env_engine.strip().lower()

        proj_engine = self.project_vars.get("engine")
        if isinstance(proj_engine, str) and proj_engine.strip():
            return proj_engine.strip().lower()

        cli_engine = self.cli_vars.get("engine")
        if isinstance(cli_engine, str) and cli_engine.strip():
            return cli_engine.strip().lower()

        return None

    def _should_register_for_engine(self, meta: Mapping[str, Any], *, path: Path) -> bool:
        """
        SQL models may declare config(engines=[...]) to limit registration.
        Returns True when the current engine matches (or no restriction given).
        """
        raw = meta.get("engines")
        if raw is None:
            return True

        tokens: Iterable[Any]
        if isinstance(raw, str):
            tokens = [raw]
        elif isinstance(raw, Iterable) and not isinstance(raw, (str, Mapping)):
            tokens = raw
        else:
            raise ModuleLoadError(
                f"{path}: config(engines=...) must be a string or iterable of strings."
            )

        allowed: set[str] = set()
        for tok in tokens:
            if not isinstance(tok, (str, bytes)):
                raise ModuleLoadError(
                    f"{path}: config(engines=...) expects strings, got {type(tok).__name__}."
                )
            text = str(tok).strip()
            if text:
                allowed.add(text.lower())

        if not allowed:
            return True

        current = self._current_engine()
        if current is None:
            raise ModuleLoadError(
                f"{path}: config(engines=...) requires an active engine.\n"
                "Hint: Export FF_ENGINE or call REGISTRY.set_active_engine('duckdb'|...)."
            )
        return current in allowed

    # def load_project(self, project_dir: Path) -> None:
    #     self.nodes.clear()
    #     self.py_funcs.clear()
    #     self.py_requires.clear()
    #     self.sources = {}
    #     self.project_vars = {}
    #     self.cli_vars = {}
    #     self.macros.clear()

    #     storage.set_model_storage({})
    #     storage.set_seed_storage({})

    #     self.project_dir = project_dir
    #     models_dir = project_dir / "models"
    #     self.env = Environment(
    #         loader=FileSystemLoader(str(models_dir)),
    #         undefined=StrictUndefined,
    #         autoescape=False,
    #         trim_blocks=True,
    #         lstrip_blocks=True,
    #     )

    #     # Make sure macros are available to all templates before model discovery.
    #     self._load_macros(models_dir)
    #     self._load_py_macros(models_dir)

    #     # load sources (version 2 schema)
    #     src_path = project_dir / "sources.yml"
    #     if src_path.exists():
    #         raw_sources = yaml.safe_load(src_path.read_text(encoding="utf-8"))
    #         try:
    #             self.sources = _parse_sources_yaml(raw_sources)
    #         except ValueError as exc:
    #             raise ValueError(f"Failed to parse sources.yml: {exc}") from exc
    #     else:
    #         self.sources = {}

    #     # load project.yml (vars)
    #     proj_path = project_dir / "project.yml"
    #     if proj_path.exists():
    #         proj_cfg = yaml.safe_load(proj_path.read_text(encoding="utf-8")) or {}
    #         self.project_vars = dict(proj_cfg.get("vars", {}) or {})

    #         models_cfg = proj_cfg.get("models") if isinstance(proj_cfg, Mapping) else None
    #         model_storage_raw = None
    #         if isinstance(models_cfg, Mapping):
    #             candidate = models_cfg.get("storage")
    #             if isinstance(candidate, Mapping):
    #                 model_storage_raw = candidate
    #         storage.set_model_storage(
    #             storage.normalize_storage_map(model_storage_raw, project_dir=project_dir)
    #         )

    #         seeds_cfg = proj_cfg.get("seeds") if isinstance(proj_cfg, Mapping) else None
    #         seed_storage_raw = None
    #         if isinstance(seeds_cfg, Mapping):
    #             candidate = seeds_cfg.get("storage")
    #             if isinstance(candidate, Mapping):
    #                 seed_storage_raw = candidate
    #         storage.set_seed_storage(
    #             storage.normalize_storage_map(seed_storage_raw, project_dir=project_dir)
    #         )

    #     # discover models
    #     for p in models_dir.rglob("*.ff.sql"):
    #         name = p.stem
    #         deps = self._scan_sql_deps(p)
    #         meta = dict(self._parse_model_config(p))
    #         storage_meta = self._lookup_storage_meta(name)
    #         if storage_meta:
    #             existing = dict(meta.get("storage") or {})
    #             existing.update(storage_meta)
    #             meta["storage"] = existing
    #         if not self._should_register_for_engine(meta, path=p):
    #             continue
    #         self._add_node_or_fail(name, "sql", p, deps, meta=meta)
    #     for p in models_dir.rglob("*.ff.py"):
    #         self._load_py_module(p)
    #         for _, func in list(self.py_funcs.items()):
    #             func_path = Path(getattr(func, "__ff_path__", "")).resolve()
    #             if func_path == p.resolve():
    #                 name = getattr(func, "__ff_name__", func.__name__)
    #                 deps = getattr(func, "__ff_deps__", [])
    #                 kind = getattr(func, "__ff_kind__", "python") or "python"

    #                 meta = dict(getattr(func, "__ff_meta__", {}) or {})
    #                 storage_meta = self._lookup_storage_meta(name)
    #                 if storage_meta:
    #                     existing = dict(meta.get("storage") or {})
    #                     existing.update(storage_meta)
    #                     meta["storage"] = existing
    #                 tags = list(getattr(func, "__ff_tags__", []) or [])
    #                 if tags:
    #                     existing_tags = meta.get("tags")
    #                     if isinstance(existing_tags, list):
    #                         merged = existing_tags + [t for t in tags if t not in existing_tags]
    #                         meta["tags"] = merged
    #                     elif existing_tags is None:
    #                         meta["tags"] = tags
    #                     else:
    #                         # Normalize non-list tags into a list while preserving the value
    #                         meta["tags"] = [existing_tags, *tags]

    #                 self._add_node_or_fail(name, kind, p, deps, meta=meta)

    #                 req = getattr(func, "__ff_require__", None)
    #                 if req:
    #                     self.py_requires[name] = req

    #     # ---- Dependency validation (early and clear)
    #     self._validate_dependencies()

    def load_project(self, project_dir: Path) -> None:
        """Load a FastFlowTransform project from the given directory."""
        self._reset_registry_state()
        self.project_dir = project_dir

        models_dir = project_dir / "models"
        self._init_jinja_env(models_dir)

        # macros first, because models may use them
        self._load_macros(models_dir)
        self._load_py_macros(models_dir)

        self._load_sources_yaml(project_dir)
        self._load_project_yaml(project_dir)

        # discover models
        self._discover_sql_models(models_dir)
        self._discover_python_models(models_dir)

        # final validation
        self._validate_dependencies()

    def _reset_registry_state(self) -> None:
        """Reset in-memory registry structures to a clean state."""
        self.nodes.clear()
        self.py_funcs.clear()
        self.py_requires.clear()
        self.sources = {}
        self.project_vars = {}
        self.cli_vars = {}
        self.macros.clear()
        # reset storage maps
        storage.set_model_storage({})
        storage.set_seed_storage({})

    def _init_jinja_env(self, models_dir: Path) -> None:
        """Initialize the Jinja environment for this project."""
        self.env = Environment(
            loader=FileSystemLoader(str(models_dir)),
            undefined=StrictUndefined,
            autoescape=False,
            trim_blocks=True,
            lstrip_blocks=True,
        )

    def _load_sources_yaml(self, project_dir: Path) -> None:
        """Load sources.yml (version 2) if present."""
        src_path = project_dir / "sources.yml"
        if not src_path.exists():
            self.sources = {}
            return

        raw_sources = yaml.safe_load(src_path.read_text(encoding="utf-8"))
        try:
            self.sources = _parse_sources_yaml(raw_sources)
        except ValueError as exc:
            raise ValueError(f"Failed to parse sources.yml: {exc}") from exc

    def _load_project_yaml(self, project_dir: Path) -> None:
        """Load project.yml (vars, storage blocks) if present."""
        proj_path = project_dir / "project.yml"
        if not proj_path.exists():
            return

        proj_cfg = yaml.safe_load(proj_path.read_text(encoding="utf-8")) or {}
        self.project_vars = dict(proj_cfg.get("vars", {}) or {})

        # models.storage
        models_cfg = proj_cfg.get("models") if isinstance(proj_cfg, Mapping) else None
        model_storage_raw = None
        if isinstance(models_cfg, Mapping):
            candidate = models_cfg.get("storage")
            if isinstance(candidate, Mapping):
                model_storage_raw = candidate
        storage.set_model_storage(
            storage.normalize_storage_map(model_storage_raw, project_dir=project_dir)
        )

        # seeds.storage
        seeds_cfg = proj_cfg.get("seeds") if isinstance(proj_cfg, Mapping) else None
        seed_storage_raw = None
        if isinstance(seeds_cfg, Mapping):
            candidate = seeds_cfg.get("storage")
            if isinstance(candidate, Mapping):
                seed_storage_raw = candidate
        storage.set_seed_storage(
            storage.normalize_storage_map(seed_storage_raw, project_dir=project_dir)
        )

    def _discover_sql_models(self, models_dir: Path) -> None:
        """Scan *.ff.sql files, parse deps, and register nodes."""
        for path in models_dir.rglob("*.ff.sql"):
            name = path.stem
            deps = self._scan_sql_deps(path)
            meta = dict(self._parse_model_config(path))
            storage_meta = self._lookup_storage_meta(name)
            if storage_meta:
                existing = dict(meta.get("storage") or {})
                existing.update(storage_meta)
                meta["storage"] = existing
            if not self._should_register_for_engine(meta, path=path):
                continue
            self._add_node_or_fail(name, "sql", path, deps, meta=meta)

    def _discover_python_models(self, models_dir: Path) -> None:
        """Scan *.ff.py files, import them, and register decorated callables."""
        for path in models_dir.rglob("*.ff.py"):
            self._load_py_module(path)

            # we might have loaded several functions; filter by file path
            for _, func in list(self.py_funcs.items()):
                func_path = Path(getattr(func, "__ff_path__", "")).resolve()
                if func_path != path.resolve():
                    continue

                name = getattr(func, "__ff_name__", func.__name__)
                deps = getattr(func, "__ff_deps__", [])
                kind = getattr(func, "__ff_kind__", "python") or "python"

                meta = dict(getattr(func, "__ff_meta__", {}) or {})
                storage_meta = self._lookup_storage_meta(name)
                if storage_meta:
                    existing = dict(meta.get("storage") or {})
                    existing.update(storage_meta)
                    meta["storage"] = existing

                # merge tags from decorator into model meta.tags
                tags = list(getattr(func, "__ff_tags__", []) or [])
                if tags:
                    existing_tags = meta.get("tags")
                    if isinstance(existing_tags, list):
                        merged = existing_tags + [t for t in tags if t not in existing_tags]
                        meta["tags"] = merged
                    elif existing_tags is None:
                        meta["tags"] = tags
                    else:
                        meta["tags"] = [existing_tags, *tags]

                self._add_node_or_fail(name, kind, path, deps, meta=meta)

                req = getattr(func, "__ff_require__", None)
                if req:
                    self.py_requires[name] = req

    # --- Macros ---------------------------------------------------------
    def _load_macros(self, models_dir: Path) -> None:
        """
        Load all Jinja macros from 'models/macros/**/*.(sql|sql.j2)' and register them
        into env.globals so they can be called directly as {{ my_macro(...) }}.
        """
        env = self.get_env()
        macros_dir = models_dir / "macros"
        if not macros_dir.exists():
            return

        files = _collect_macro_files(macros_dir)
        if not files:
            return

        for path in files:
            rel = _relative_name(path, models_dir)
            tmpl = _get_or_build_template(env, path, rel)
            mod = _template_module_or_none(tmpl)
            if mod is None:
                continue

            for name, obj in _iter_public_attrs(mod):
                if _is_jinja_macro(obj):
                    env.globals[name] = obj  # last-one-wins ok
                    self.macros[name] = path

    def _load_py_macros(self, models_dir: Path) -> None:
        """
        Load Python helpers from 'models/macros_py/**/*.py' and register all public
        callables as Jinja globals & filters.
        """
        env = self.get_env()
        py_dir = models_dir / "macros_py"
        if not py_dir.exists():
            return

        for p in sorted(py_dir.rglob("*.py")):
            # unique module name to avoid caching collisions across tests/runs
            mod_name = f"ff_macros_{p.stem}_{abs(hash(str(p.resolve()))):x}"

            spec = importlib.util.spec_from_file_location(mod_name, p)
            if not spec or not spec.loader:
                continue

            mod = importlib.util.module_from_spec(spec)
            try:
                spec.loader.exec_module(mod)  # executes user code
            except Exception as e:
                # In Tests willst du das sehen; wenn du es leise ignorieren willst -> 'continue'
                raise RuntimeError(f"Failed to import macro helper {p}: {e}") from e

            for name, obj in vars(mod).items():
                if name.startswith("_") or not callable(obj):
                    continue
                env.globals[name] = obj
                with suppress(Exception):
                    env.filters[name] = obj
                self.macros[name] = p

    def _load_py_module(self, path: Path) -> types.ModuleType:
        """
        Load a Python module from filesystem path in a typing-safe way.
        Ensures both spec and spec.loader are non-None, otherwise raises.
        """
        # Important: use absolute paths so later comparisons work
        path = path.resolve()

        spec = importlib.util.spec_from_file_location(path.stem, path)
        if spec is None:
            raise ModuleLoadError(f"Unable to create module spec for {path}")

        if spec.loader is None:
            raise ModuleLoadError(f"Module spec has no loader for {path}")

        mod = importlib.util.module_from_spec(spec)
        # exec_module is part of the loader protocol; Pylance now knows the type
        spec.loader.exec_module(mod)
        return mod

    def _add_node_or_fail(
        self, name: str, kind: str, path: Path, deps: list[str], *, meta: dict[str, Any]
    ) -> None:
        if name in self.nodes:
            other = self.nodes[name].path
            raise ModuleLoadError(
                "Duplicate model name detected:\n"
                f"• alredy registered: {other}\n"
                f"• new model:        {path}\n"
                "Hint: Rename one of the models (file name = node name)"
                "or use @model(name='…') for Python."
            )
        self.nodes[name] = Node(name=name, kind=kind, path=path, deps=deps, meta=meta)

    def _scan_sql_deps(self, path: Path) -> list[str]:
        txt = path.read_text(encoding="utf-8")
        literal = re.compile(r"ref\s*\(\s*['\"]([A-Za-z0-9_.\-]+)['\"]\s*\)")
        dynamic = re.compile(r"ref\s*\(\s*([^)]+)\)")

        deps = literal.findall(txt)

        for expr in dynamic.findall(txt):
            expr_stripped = expr.strip()
            if not (
                (expr_stripped.startswith("'") and expr_stripped.endswith("'"))
                or (expr_stripped.startswith('"') and expr_stripped.endswith('"'))
            ):
                logger = get_logger("registry")
                logger.warning(
                    "%s: ref(%s) cannot be statically resolved; DAG may miss this dependency. "
                    "Wrap options in a mapping of literal ref('...') calls and pick from that map.",
                    path,
                    expr_stripped,
                )

        return deps

    # -------- {{ config(...) }} Head-Parser --------
    def _parse_model_config(self, path: Path) -> dict[str, Any]:
        """
        Reads the leading line {{ config(materialized='view', key=1) }}.
        Safely parses via ast.literal_eval for keyword arguments. Errors → {}.
        """
        try:
            head = path.read_text(encoding="utf-8", errors="ignore")[:2000]
        except Exception:
            return {}
        m = re.search(
            r"^\s*\{\{\s*config\s*\((?P<args>.*?)\)\s*\}\}", head, flags=re.IGNORECASE | re.DOTALL
        )
        if not m:
            return {}
        args = m.group("args").strip()
        if not args:
            return {}
        try:
            # parse "a=1, b='x'" as a Call and extract keywords
            node = ast.parse(f"__CFG__({args})", mode="eval")
            if not isinstance(node.body, ast.Call):
                return {}
            cfg: dict[str, Any] = {}
            for kw in node.body.keywords:
                if kw.arg is None:
                    # **kwargs werden (noch) ignoriert
                    continue
                cfg[kw.arg] = ast.literal_eval(kw.value)
            return cfg
        except Exception:
            # Robust: keine Hard-Fails beim Laden
            return {}

    def _validate_dependencies(self) -> None:
        """
        Collect all missing dependencies across nodes and raise
        DependencyNotFoundError with a precise list and hints.
        """
        missing_map: dict[str, list[str]] = {}
        known = set(self.nodes.keys())
        for node in self.nodes.values():
            # Only validate actual model refs - source() targets are not nodes
            missing = [dep for dep in (node.deps or []) if dep not in known]
            if missing:
                missing_map[node.name] = missing

        if missing_map:
            raise DependencyNotFoundError(missing_map)

get_project_dir

get_project_dir()

Return the project directory after load_project(), or raise if not set.

Source code in src/fastflowtransform/core.py

def get_project_dir(self) -> Path:
    """Return the project directory after load_project(), or raise if not set."""
    if self.project_dir is None:
        raise RuntimeError("Project directory not initialized. Call load_project() first.")
    return self.project_dir

get_env

get_env()

Return the initialized Jinja Environment, or raise if not loaded.

Source code in src/fastflowtransform/core.py

def get_env(self) -> Environment:
    """Return the initialized Jinja Environment, or raise if not loaded."""
    if self.env is None:
        raise RuntimeError("Jinja environment not initialized. Call load_project() first.")
    return self.env

set_cli_vars

set_cli_vars(overrides)

Set CLI --vars overrides (highest precedence).

Source code in src/fastflowtransform/core.py

def set_cli_vars(self, overrides: dict[str, Any]) -> None:
    """Set CLI --vars overrides (highest precedence)."""
    self.cli_vars = dict(overrides or {})

set_active_engine

set_active_engine(engine)

Store active engine hint (case-insensitive) for conditional loading.

Source code in src/fastflowtransform/core.py

def set_active_engine(self, engine: str | None) -> None:
    """Store active engine hint (case-insensitive) for conditional loading."""
    self.active_engine = engine.lower().strip() if isinstance(engine, str) else None

load_project

load_project(project_dir)

Load a FastFlowTransform project from the given directory.

Source code in src/fastflowtransform/core.py

def load_project(self, project_dir: Path) -> None:
    """Load a FastFlowTransform project from the given directory."""
    self._reset_registry_state()
    self.project_dir = project_dir

    models_dir = project_dir / "models"
    self._init_jinja_env(models_dir)

    # macros first, because models may use them
    self._load_macros(models_dir)
    self._load_py_macros(models_dir)

    self._load_sources_yaml(project_dir)
    self._load_project_yaml(project_dir)

    # discover models
    self._discover_sql_models(models_dir)
    self._discover_python_models(models_dir)

    # final validation
    self._validate_dependencies()

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