feat(tangle-cli): add python_pipeline authoring DSL

packages/tangle-cli/src/tangle_cli/component_from_func.py

@@ -757,13 +757,22 @@ def _is_main_str(n: ast.expr) -> bool:
 # too, exactly like @task.
 _AUTHORING_DECORATOR_NAMES = frozenset({"task", "pipeline", "subpipeline", "registered"})
 
-# The python-pipeline authoring module. ONLY imports of this module (and its
-# submodules) are authoring-only and stripped from the baked source. We
-# deliberately do NOT strip other ``tangle_deploy.*`` packages (e.g.
-# ``tangle_deploy.utils``): those may be legitimate runtime helpers used inside a
-# ``@task`` body, and dropping them would raise ``NameError`` in the operation
-# container.
-_AUTHORING_IMPORT_MODULE = "tangle_deploy.python_pipeline"
+# The python-pipeline authoring modules. ONLY imports of these modules (and
+# their submodules) are authoring-only and stripped from the baked source. We
+# deliberately do NOT strip other ``tangle_deploy.*`` / ``tangle_cli.*`` packages
+# (e.g. ``tangle_deploy.utils``): those may be legitimate runtime helpers used
+# inside a ``@task`` body, and dropping them would raise ``NameError`` in the
+# operation container.
+#
+# Two authoring surfaces are recognised: the legacy ``tangle_deploy.python_pipeline``
+# path (used by every pipeline authored before the OSS move) and the canonical
+# OSS ``tangle_cli.python_pipeline`` path. Both expose the identical decorators —
+# ``tangle_deploy.python_pipeline`` re-exports the OSS objects — so an author may
+# use either import and codegen must strip either the same way.
+_AUTHORING_IMPORT_MODULES = (
+    "tangle_deploy.python_pipeline",
+    "tangle_cli.python_pipeline",
+)
 
 # The authoring-only ``TaskEnv`` class name. A module-level ``X = TaskEnv(...)``
 # (or ``X = <alias>.TaskEnv(...)``) declaration is authoring-only by contract and
@@ -812,33 +821,35 @@ def _decorator_called_name(node: ast.expr) -> str | None:
     return None
 
 
+def _is_authoring_module(name: str) -> bool:
+    """Return True if *name* is an authoring module or a submodule of one."""
+    return any(name == mod or name.startswith(mod + ".") for mod in _AUTHORING_IMPORT_MODULES)
+
+
 def _is_authoring_import(node: ast.stmt) -> bool:
     """Return True if *node* imports the python-pipeline authoring surface.
 
-    Matches ONLY the ``tangle_deploy.python_pipeline`` module (and its
-    submodules):
+    Matches ONLY the ``tangle_deploy.python_pipeline`` / ``tangle_cli.python_pipeline``
+    modules (and their submodules):
 
-    - ``from tangle_deploy.python_pipeline import ...`` (including the aliased
-      ``from tangle_deploy.python_pipeline import ref as operation_by_ref`` form
-      and submodules like ``from tangle_deploy.python_pipeline.x import y``);
-    - ``import tangle_deploy.python_pipeline`` / ``import
-      tangle_deploy.python_pipeline as tp``.
+    - ``from tangle_cli.python_pipeline import ...`` (including the aliased
+      ``from tangle_cli.python_pipeline import ref as operation_by_ref`` form
+      and submodules like ``from tangle_cli.python_pipeline.x import y``);
+    - ``import tangle_cli.python_pipeline`` / ``import
+      tangle_cli.python_pipeline as tp``;
+    - the legacy ``tangle_deploy.python_pipeline`` equivalents.
 
-    It does NOT match other ``tangle_deploy.*`` packages (e.g.
+    It does NOT match other ``tangle_deploy.*`` / ``tangle_cli.*`` packages (e.g.
     ``from tangle_deploy.utils import X``) — those can be genuine runtime helpers
     referenced inside a ``@task`` body and must survive into the baked program.
     Relative imports (``from . import x``) are never authoring imports.
     """
     if isinstance(node, ast.ImportFrom):
         if node.level:  # relative import — not the authoring package
             return False
-        module = node.module or ""
-        return module == _AUTHORING_IMPORT_MODULE or module.startswith(_AUTHORING_IMPORT_MODULE + ".")
+        return _is_authoring_module(node.module or "")
     if isinstance(node, ast.Import):
-        return any(
-            alias.name == _AUTHORING_IMPORT_MODULE or alias.name.startswith(_AUTHORING_IMPORT_MODULE + ".")
-            for alias in node.names
-        )
+        return any(_is_authoring_module(alias.name) for alias in node.names)
     return False
 
 

packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py

@@ -0,0 +1,43 @@
+"""Python-first authoring surface for Tangle pipelines.
+
+End users write::
+
+    from tangle_cli.python_pipeline import pipeline, task, registered, ref, raw, subpipeline, TaskEnv, In, Out
+
+``cfg`` is NOT a top-level export — it is a parameter the framework
+injects into the user's pipeline function at trace time. Importing the
+:class:`tangle_cli.python_pipeline.cfg.Cfg` class is reserved for the
+compile driver.
+
+``import tangle_cli.python_pipeline`` is kept light: it does not
+eagerly import the heavy ``tangle_cli.component_generator`` codegen
+module or the tracer machinery.
+
+Module map: authoring entry points live in :mod:`.pipeline`,
+:mod:`.task`, :mod:`.subpipeline`, :mod:`.registered`, :mod:`.ref` and
+:mod:`.raw`; the trace-time IR is built in :mod:`.trace` / :mod:`.graph`
+and lowered to the dehydrated dict shape by :mod:`.emit`.
+"""
+from __future__ import annotations
+
+from .pipeline import pipeline
+from .raw import raw
+from .ref import ref
+from .registered import registered
+from .subpipeline import subpipeline
+from .task import task
+from .task_env import TaskEnv
+from .types import In, Out, Outputs
+
+__all__ = [
+    "pipeline",
+    "task",
+    "registered",
+    "ref",
+    "raw",
+    "subpipeline",
+    "TaskEnv",
+    "In",
+    "Out",
+    "Outputs",
+]

packages/tangle-cli/src/tangle_cli/python_pipeline/cfg.py

@@ -0,0 +1,169 @@
+"""Config loader: YAML → attribute-access object.
+
+``Cfg`` is a pure, operation-agnostic attribute-access wrapper. It holds
+NO operation-specific knowledge (no SQL, BigQuery table FQN building, or
+other domain helpers): a pipeline that needs a fully-qualified BigQuery
+table name (or any other domain-specific value) builds it in its own
+pipeline-local helper code and passes the final string in as a constant.
+
+Contract:
+
+- ``cfg.<flat_key>`` returns the value (flat keys win over nested).
+- ``cfg.<a>.<b>`` returns the nested value if ``a`` maps to a dict.
+- Unknown keys raise :class:`UnknownCfgKeyError`.
+- ``--override key=value`` pairs overlay file values (CLI wins).
+- ``--override`` *values* are typed via ``yaml.safe_load`` (parity with
+  file values, which already pass through ``yaml.safe_load``): so
+  ``--override batch_size=100`` yields ``int 100`` (not ``"100"``) and
+  ``--override dry_run=false`` yields ``bool False`` (not the truthy
+  string ``"false"``). The one special case is an EMPTY value: ``key=``
+  maps to ``""`` (an empty string), NOT ``None`` — a bare ``--override
+  key=`` reads as "set to empty string". Authors who want an explicit
+  null write ``--override key=null``. Because the same loader the config
+  file uses is reused, the YAML 1.1 quirks are inherited and CONSISTENT
+  with file values (e.g. ``yes/no/on/off`` -> bool, ``3:14`` -> 194,
+  ``007`` -> 7, ``0x1F`` -> 31, ``1_000`` -> 1000, ``1.0`` -> float; but
+  ``1e3`` stays a string). Quote to force a string: ``--override
+  version='"1.0"'``.
+- ``template_file:`` in the source is rejected — the Python authoring
+  layer IS the template authoring layer.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from .errors import CompileError, UnknownCfgKeyError
+
+
+def _coerce_override(value: str) -> Any:
+    """Type a single ``--override key=value`` *value* string.
+
+    Override values arrive from the CLI as raw strings (see
+    ``pipeline_compiler._parse_overrides``). To give an override the SAME
+    Python type the equivalent ``key: value`` line in ``config.yaml`` would
+    (which is already typed via ``yaml.safe_load``), we run each value
+    through the same loader.
+
+    The one deviation from raw YAML is the EMPTY string: ``yaml.safe_load("")``
+    is ``None``, but a bare ``--override key=`` reads as "set to empty
+    string", so we keep ``""`` as ``""``. (An explicit null is available via
+    ``--override key=null``.)
+    """
+    if value == "":
+        return ""
+    return yaml.safe_load(value)
+
+
+class _CfgNested:
+    """A nested mapping inside ``Cfg`` — same attribute-access semantics."""
+
+    def __init__(self, data: dict[str, Any]) -> None:
+        object.__setattr__(self, "_data", data)
+
+    def __getattr__(self, key: str) -> Any:
+        data: dict[str, Any] = object.__getattribute__(self, "_data")
+        if key not in data:
+            raise UnknownCfgKeyError(f"unknown config key: {key}")
+        value = data[key]
+        if isinstance(value, dict):
+            return _CfgNested(value)
+        return value
+
+
+class Cfg:
+    """Attribute-access wrapper around a flat-ish config dict.
+
+    Read-only at runtime; immutability is not enforced (Python attribute
+    semantics) but writes are not part of the public API.
+    """
+
+    def __init__(self, data: dict[str, Any]) -> None:
+        # Use object.__setattr__ to avoid recursing through __getattr__.
+        object.__setattr__(self, "_data", dict(data))
+
+    # ------------------------------------------------------------------
+    # Public attribute-access surface
+
+    def __getattr__(self, key: str) -> Any:
+        data: dict[str, Any] = object.__getattribute__(self, "_data")
+        if key in data:
+            value = data[key]
+            if isinstance(value, dict):
+                return _CfgNested(value)
+            return value
+        raise UnknownCfgKeyError(f"unknown config key: {key!r}. Available keys: " f"{sorted(data.keys())!r}")
+
+
+def load_cfg(path: Path, overrides: Mapping[str, Any] | None = None, *, coerce: bool = True) -> Cfg:
+    """Load YAML at ``path`` and overlay ``overrides`` on top.
+
+    Two overlay paths, selected by ``coerce``:
+
+    * ``coerce=True`` (the default, the CLI ``--override`` path): override
+      *values* are raw strings typed via :func:`_coerce_override` (i.e.
+      ``yaml.safe_load``), so an override yields the SAME Python type the
+      equivalent line in ``config.yaml`` would: ``batch_size=100`` ->
+      ``int 100``, ``dry_run=false`` -> ``bool False``, ``ratio=0.1`` ->
+      ``float 0.1``, ``name=foo`` -> ``"foo"``. The empty value ``key=`` is
+      special-cased to ``""`` (not ``None``); ``key=null`` -> ``None``. The
+      same YAML 1.1 quirks the file loader has are inherited (and thus
+      consistent); quote to force a string (``--override version='"1.0"'``).
+    * ``coerce=False`` (the NATIVE pass-through path, used for programmatic
+      ``.override_config`` / broadcast values): override values are
+      already-typed Python natives and are overlaid AS-IS, NOT re-coerced
+      through ``yaml.safe_load``. Re-coercion is correct only for raw CLI
+      strings — it would, e.g., turn the string ``"no"`` into ``False`` or
+      fail on a non-string value.
+
+    See the module docstring for the full coercion contract.
+
+    Args:
+        path: Path to the user's config.yaml.
+        overrides: Optional mapping of override pairs. Under ``coerce=True``
+            these are raw ``--override key=value`` strings; under
+            ``coerce=False`` they are already-typed native values. CLI /
+            override wins on conflict.
+        coerce: When ``True`` (default), values are coerced to their YAML
+            type before overlaying. When ``False``, values pass through
+            unchanged (native overlay).
+
+    Returns:
+        A :class:`Cfg` ready for attribute access.
+
+    Raises:
+        CompileError: if the config carries a ``template_file:`` key
+            (the authoring layer emits the template itself; the source
+            config must not point at a separate Jinja template).
+        FileNotFoundError: if ``path`` doesn't exist.
+    """
+    raw = yaml.safe_load(Path(path).read_text()) or {}
+    if not isinstance(raw, dict):
+        raise CompileError(f"config at {path} must be a YAML mapping, got {type(raw).__name__}")
+    if "template_file" in raw:
+        raise CompileError(
+            f"config at {path} has a top-level `template_file:` key, but "
+            "the Python authoring layer emits the pipeline directly. Remove "
+            "this key from your config.yaml — the framework is your "
+            "templating layer."
+        )
+    merged = dict(raw)
+    if overrides:
+        if coerce:
+            # Coerce each raw CLI override string to its YAML type so an
+            # override behaves identically to the same key written in the
+            # config file (which is already typed via yaml.safe_load above).
+            # Coercion is internal to load_cfg only: the override dict stays
+            # dict[str, str] everywhere upstream and the compile-cache
+            # fingerprint keeps hashing the raw CLI strings.
+            merged.update({k: _coerce_override(v) for k, v in overrides.items()})
+        else:
+            # Native pass-through: values are already-typed Python natives
+            # (from .override_config / broadcast). Overlay verbatim — no
+            # yaml.safe_load re-coercion (it would mangle e.g. "no" -> False).
+            merged.update(dict(overrides))
+    return Cfg(merged)