Feature: metadata helpers for pipeline integration (tested for egon-data)

src/omi/cli.py

@@ -623,6 +623,103 @@ def inspect_db_drift_cmd(  # noqa: PLR0913
         click.secho(f"✓ Automatically updated resource YAML: {resource_path.name}", fg="green")
 
 
+@grp.command("add-source")
+@click.argument("resource_yaml", type=click.Path(exists=True, dir_okay=False, path_type=Path))
+@click.option(
+    "--kind",
+    type=click.Choice(["external", "internal"]),
+    default="external",
+    show_default=True,
+    help="external: literature/other database; internal: another table in this project.",
+)
+@click.option("--title", default=None, help="Source title (required for --kind external).")
+@click.option("--table", "ref_table", default=None, help="Referenced table name (required for --kind internal).")
+@click.option("--path", "src_path", default="", help="URL or reference path of the source.")
+@click.option("--description", default="", help="Human-readable source description.")
+def add_source_cmd(  # noqa: PLR0913
+    resource_yaml: Path,
+    kind: str,
+    title: Optional[str],
+    ref_table: Optional[str],
+    src_path: str,
+    description: str,
+) -> None:
+    """Append a provenance source to a resource YAML (non-destructive, de-duplicated)."""
+    from omi.creation.sources import (
+        add_source_to_resource_file,
+        build_external_source,
+        build_internal_source,
+    )
+
+    if kind == "internal":
+        if not ref_table:
+            msg = "--table is required for --kind internal"
+            raise click.UsageError(msg)
+        source = build_internal_source(ref_table, title=title, description=description, path=src_path)
+    else:
+        if not title:
+            msg = "--title is required for --kind external"
+            raise click.UsageError(msg)
+        source = build_external_source(title, description=description, path=src_path)
+
+    added = add_source_to_resource_file(resource_yaml, source)
+    if added:
+        click.secho(f"✓ Added {kind} source to {resource_yaml}", fg="green")
+    else:
+        click.secho(f"= Equivalent source already present in {resource_yaml}; unchanged", fg="yellow")
+
+
+@grp.command("coverage")
+@click.argument("base_dir", type=click.Path(exists=True, file_okay=False, path_type=Path))
+@click.option(
+    "--expected-file",
+    required=True,
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    help="File listing expected resource names, one per line ('#' comments allowed).",
+)
+@click.option("--dataset-id", "dataset_ids", multiple=True, help="Restrict scan to these dataset ids (repeatable).")
+@click.option("--strict", is_flag=True, help="Exit non-zero if any expected resource is missing.")
+@click.option(
+    "--no-field-descriptions",
+    is_flag=True,
+    help="Do not require per-field descriptions for a resource to count as complete.",
+)
+def coverage_cmd(
+    base_dir: Path,
+    expected_file: Path,
+    dataset_ids: tuple[str, ...],
+    *,
+    strict: bool,
+    no_field_descriptions: bool,
+) -> None:
+    """Report documented/skeleton/missing/orphan coverage against an expected resource list."""
+    from omi.creation.coverage import coverage_report
+
+    expected = [
+        line.strip()
+        for line in expected_file.read_text(encoding="utf-8").splitlines()
+        if line.strip() and not line.strip().startswith("#")
+    ]
+    report = coverage_report(
+        base_dir,
+        expected,
+        dataset_ids=dataset_ids or None,
+        require_field_descriptions=not no_field_descriptions,
+    )
+
+    click.echo(report.summary())
+    for name in report.missing:
+        click.secho(f"[missing]  {name}", fg="red")
+    for state in report.skeleton:
+        detail = "; ".join(state.reasons[:3])
+        click.secho(f"[skeleton] {state.name}  ({detail})", fg="yellow")
+    for name in report.orphans:
+        click.secho(f"[orphan]   {name}", fg="cyan")
+
+    if strict and not report.ok:
+        raise click.Abort
+
+
 # Keep CommandCollection for backwards compatibility with your entry point
 cli = click.CommandCollection(sources=[grp, init, inspect])
 

src/omi/conversions/v160_to_v20.py

@@ -304,23 +304,48 @@ def rename_path_to_id(annotation_object: dict) -> dict:
 
 
 def ___v2_populate_schema_primary_keys(resource_v2: dict, resource: dict) -> None:
-    """Populate schema fields in resource_v2 from resource in v1.6."""
-    for i_pk, pk in enumerate(resource.get("schema", {}).get("primaryKey", []) or []):
-        if i_pk >= len(resource_v2["schema"]["primaryKey"]):
-            resource_v2["schema"]["primaryKey"].append(deepcopy(resource_v2["schema"]["primaryKey"][0]))
+    """
+    Populate schema.primaryKey in resource_v2 from the v1.6 resource.
+
+    v1.5/v1.6 may store ``primaryKey`` as a comma-separated string
+    (e.g. ``"id, scn_name"``) or as a list; OEMetadata v2 requires a list of
+    column names. A string is split on commas and whitespace-stripped; the
+    ``"none"`` sentinel and empty values become an empty list. Iterating the
+    string directly (the previous behaviour) split it into individual
+    characters, so this coercion must happen before any iteration.
+    """
+    raw_pk = resource.get("schema", {}).get("primaryKey", []) or []
 
-        if isinstance(pk, str):
-            resource_v2["schema"]["primaryKey"].pop()
-            resource_v2["schema"]["primaryKey"].append(pk)
+    if isinstance(raw_pk, str):
+        if raw_pk.strip().lower() in ("", "none"):
+            primary_keys: list = []
+        else:
+            primary_keys = [part.strip() for part in raw_pk.split(",") if part.strip()]
+    else:
+        primary_keys = [pk for pk in raw_pk if pk not in (None, "")]
+
+    resource_v2["schema"]["primaryKey"] = primary_keys
 
 
 def ___v2_populate_schema_foreign_keys(resource_v2: dict, resource: dict) -> None:
-    """Populate schema fields in resource_v2 from resource in v1.6."""
-    for i_fk, fk in enumerate(resource.get("schema", {}).get("foreignKeys", [])):
+    """
+    Populate schema.foreignKeys in resource_v2 from the v1.6 resource.
+
+    When the source declares no foreign keys, emit an empty list rather than
+    leaking the v2 template placeholder stub (an empty ``{fields, reference}``
+    entry) into the output.
+    """
+    source_fks = resource.get("schema", {}).get("foreignKeys", []) or []
+
+    if not source_fks:
+        resource_v2["schema"]["foreignKeys"] = []
+        return
+
+    for i_fk, fk in enumerate(source_fks):
         if i_fk >= len(resource_v2["schema"]["foreignKeys"]):
             resource_v2["schema"]["foreignKeys"].append(deepcopy(resource_v2["schema"]["foreignKeys"][0]))
 
-        if isinstance(fk, object):
+        if isinstance(fk, dict):
             resource_v2["schema"]["foreignKeys"][i_fk].update(
                 (k, resource["schema"]["foreignKeys"][i_fk][k])
                 for k in resource_v2["schema"]["foreignKeys"][i_fk].keys()

src/omi/creation/coverage.py

@@ -0,0 +1,200 @@
+# omi/creation/coverage.py
+"""
+Coverage reporting for an OEMetadata split-YAML store.
+
+Given an *expected* set of resource names (the tables that should be
+documented -- supplied by the caller, e.g. derived from a pipeline's declared
+outputs), report which are:
+
+- **missing**  -- expected but no resource YAML exists (hard gap),
+- **skeleton** -- a YAML exists but required human fields are still blank or
+  carry scaffolding placeholders (soft gap / warning),
+- **complete** -- fully documented,
+
+plus **orphans** -- resources present in the store that nobody expects.
+
+The expected list is caller-supplied so this stays project-agnostic: OMI does
+not know about any particular pipeline's output declarations; the caller
+computes the list and passes resource names in.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Optional, Union
+
+from .utils import discover_dataset_ids, discover_paths, load_yaml
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+Json = dict[str, object]
+
+# a filled field must not still carry a scaffolding placeholder
+_PLACEHOLDER_MARKERS: tuple[str, ...] = ("TODO", "WILL_BE_SET_AT_PUBLICATION")
+# resource-level fields a human must fill for a resource to count as complete
+_DEFAULT_REQUIRED_RESOURCE_FIELDS: tuple[str, ...] = ("title", "description")
+
+
+@dataclass
+class ResourceState:
+    """Coverage state of a single expected resource."""
+
+    name: str
+    state: str  # "missing" | "skeleton" | "complete"
+    path: Optional[Path] = None
+    reasons: list[str] = field(default_factory=list)
+
+
+@dataclass
+class CoverageReport:
+    """Result of comparing an expected resource list against the store."""
+
+    missing: list[str] = field(default_factory=list)  # expected, no YAML
+    skeleton: list[ResourceState] = field(default_factory=list)
+    complete: list[str] = field(default_factory=list)
+    orphans: list[str] = field(default_factory=list)  # in store, not expected
+    states: dict[str, ResourceState] = field(default_factory=dict)
+
+    @property
+    def ok(self) -> bool:
+        """True when nothing expected is missing (skeletons are warnings)."""
+        return not self.missing
+
+    def summary(self) -> str:
+        """One-line human-readable tally."""
+        return (
+            f"complete={len(self.complete)} "
+            f"skeleton={len(self.skeleton)} "
+            f"missing={len(self.missing)} "
+            f"orphans={len(self.orphans)}"
+        )
+
+
+def _is_blank(value: object) -> bool:
+    return value in (None, "", [], {})
+
+
+def _has_placeholder(value: object) -> bool:
+    return isinstance(value, str) and any(m in value for m in _PLACEHOLDER_MARKERS)
+
+
+def resource_completeness(
+    resource: Json,
+    *,
+    required_fields: Iterable[str] = _DEFAULT_REQUIRED_RESOURCE_FIELDS,
+    require_field_descriptions: bool = True,
+) -> tuple[str, list[str]]:
+    """
+    Classify a resource dict as ``"complete"`` or ``"skeleton"``.
+
+    A resource is *complete* when every required resource-level field is
+    non-empty and free of scaffolding placeholders and -- when
+    ``require_field_descriptions`` is set -- every schema field carries a
+    placeholder-free description. Otherwise it is a *skeleton*.
+
+    Returns
+    -------
+    (state, reasons)
+        ``reasons`` lists every unmet condition (empty when complete).
+    """
+    reasons: list[str] = []
+
+    for name in required_fields:
+        value = resource.get(name)
+        if _is_blank(value):
+            reasons.append(f"'{name}' is empty")
+        elif _has_placeholder(value):
+            reasons.append(f"'{name}' still has a placeholder")
+
+    if require_field_descriptions:
+        schema = resource.get("schema")
+        fields = schema.get("fields") if isinstance(schema, dict) else None
+        if isinstance(fields, list):
+            for fld in fields:
+                if not isinstance(fld, dict):
+                    continue
+                col = fld.get("name", "?")
+                desc = fld.get("description")
+                if _is_blank(desc):
+                    reasons.append(f"field '{col}' has no description")
+                elif _has_placeholder(desc):
+                    reasons.append(f"field '{col}' description is a placeholder")
+
+    return ("complete" if not reasons else "skeleton", reasons)
+
+
+def index_store_resources(
+    base_dir: Union[str, Path],
+    *,
+    dataset_ids: Optional[Iterable[str]] = None,
+) -> dict[str, Path]:
+    """
+    Map every resource *name* present in the store to its YAML path.
+
+    Scans the given dataset ids (or all discovered ids) and reads each
+    resource YAML's ``name`` field. A name defined in more than one dataset
+    resolves to the last one scanned.
+    """
+    base = Path(base_dir)
+    ids = list(dataset_ids) if dataset_ids is not None else discover_dataset_ids(base)
+    index: dict[str, Path] = {}
+    for ds_id in ids:
+        _, _, resource_paths = discover_paths(base, ds_id)
+        for path in resource_paths:
+            doc = load_yaml(path)
+            name = doc.get("name")
+            if isinstance(name, str) and name:
+                index[name] = path
+    return index
+
+
+def coverage_report(
+    base_dir: Union[str, Path],
+    expected: Iterable[str],
+    *,
+    dataset_ids: Optional[Iterable[str]] = None,
+    required_fields: Iterable[str] = _DEFAULT_REQUIRED_RESOURCE_FIELDS,
+    require_field_descriptions: bool = True,
+) -> CoverageReport:
+    """
+    Compare an expected resource-name list against the store.
+
+    Parameters
+    ----------
+    base_dir :
+        Path to the metadata store (contains ``datasets/`` and ``resources/``).
+    expected :
+        Resource names that *should* be documented (e.g. ``schema.table``).
+    dataset_ids :
+        Restrict the scan to these dataset ids (default: all discovered).
+    required_fields, require_field_descriptions :
+        Passed through to :func:`resource_completeness` to tune the
+        skeleton/complete threshold.
+    """
+    expected_names = list(dict.fromkeys(expected))  # dedupe, preserve order
+    store = index_store_resources(base_dir, dataset_ids=dataset_ids)
+    required = tuple(required_fields)
+
+    report = CoverageReport()
+    for name in expected_names:
+        path = store.get(name)
+        if path is None:
+            state = ResourceState(name=name, state="missing")
+            report.missing.append(name)
+        else:
+            level, reasons = resource_completeness(
+                load_yaml(path),
+                required_fields=required,
+                require_field_descriptions=require_field_descriptions,
+            )
+            state = ResourceState(name=name, state=level, path=path, reasons=reasons)
+            if level == "complete":
+                report.complete.append(name)
+            else:
+                report.skeleton.append(state)
+        report.states[name] = state
+
+    expected_set = set(expected_names)
+    report.orphans = sorted(name for name in store if name not in expected_set)
+    return report