feat(python): MetaDataLoader class with from_directory/from_uris/from_string factories

Replaces the free `load_directory` function with a source-polymorphic
MetaDataLoader class — the Tier-1 invariant API matching the other three
ports. `loader.load(sources)` dispatches per-source on declared format
(json -> parse_document; yaml -> parse_yaml), so callers can mix file /
directory / URI / in-memory sources in one call.

Three class-method factories cover the 99% case:
  - MetaDataLoader.from_directory(dir, recurse=True)
  - MetaDataLoader.from_uris([...])
  - MetaDataLoader.from_string(content, format=JSON)

Existing callsites still import `load_directory` from this module and will
break in collection. They are migrated to the new API in the next commit
(Task 4.4); the cross-port `__init__` shortcuts arrive in Task 4.3.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dmealing

server/python/src/metaobjects/loader/meta_data_loader.py

@@ -1,4 +1,17 @@
-"""Filesystem loader: discover -> parse -> merge -> freeze."""
+"""MetaDataLoader: source-polymorphic loader.
+
+The cross-language Tier-1 loader: takes a list of MetaDataSource impls,
+runs parse -> merge -> super-resolve -> validate -> freeze, and returns a
+LoadResult { root, errors, warnings }.
+
+Source format (JSON vs YAML) is declared by the source, not sniffed by the
+loader. This replaces both the per-file extension switch from the old
+free function and the TS FileMetaDataLoader.parseSource override.
+
+The three class-method factories (from_directory / from_uris / from_string)
+cover the 99% case; module-level shortcuts in `metaobjects.__init__` wrap
+them with Pythonic ergonomics.
+"""
 from __future__ import annotations
 
 import json
@@ -16,74 +29,138 @@
 from ..shared.base_types import SUBTYPE_ROOT, TYPE_METADATA
 from ..super_resolve import resolve_supers
 from .merge import merge_roots
+from .sources import (
+    DirectorySource,
+    InMemoryStringSource,
+    MetaDataFormat,
+    MetaDataSource,
+    UriSource,
+)
 from .validation_passes import run_validations
 
-# Authoring file formats this loader recognises. JSON is canonical interchange;
-# YAML is the sigil-free authoring front-end (ADR-0006).
-_JSON_SUFFIXES = (".json",)
-_YAML_SUFFIXES = (".yaml", ".yml")
-
 
 @dataclass
 class LoadResult:
+    """The Tier-1 load result. Same field shape across all four ports."""
+
     root: MetaData
     errors: list[MetaError] = field(default_factory=list)
     warnings: list[str] = field(default_factory=list)
 
 
-def _parse_file(path: Path, registry: TypeRegistry, errors: list[MetaError]) -> ParseResult | None:
-    """Dispatch by file extension to the JSON or YAML front-end.
+def _empty_root() -> MetaRoot:
+    """Construct the canonical empty root (matches parse_document's seed)."""
+    return MetaRoot(TYPE_METADATA, SUBTYPE_ROOT, "")
+
+
+class MetaDataLoader:
+    """Source-polymorphic metadata loader.
 
-    Returns None if the file could not even be read/parsed at the syntax level
-    (e.g. malformed JSON/YAML); the caller adds the appropriate error to
-    *errors* and skips this file from the merge set.
+    Construct once with a provider list (defaults to ``[core_provider]``);
+    call ``.load(sources)`` for arbitrary source combinations, or use the
+    ``from_*`` class-method factories for the common cases.
     """
-    suffix = path.suffix.lower()
-    text = path.read_text(encoding="utf-8-sig")
-    if suffix in _YAML_SUFFIXES:
+
+    def __init__(self, providers: list[Provider] | None = None) -> None:
+        self._registry: TypeRegistry = compose_registry(
+            providers if providers is not None else [core_provider]
+        )
+
+    @property
+    def registry(self) -> TypeRegistry:
+        return self._registry
+
+    # --- core API ------------------------------------------------------
+
+    def load(self, sources: list[MetaDataSource]) -> LoadResult:
+        """Parse -> merge -> super-resolve -> validate -> freeze."""
+        result = LoadResult(root=_empty_root())
+        roots: list[MetaData] = []
+
+        for src in sources:
+            parsed = self._parse_source(src, result.errors)
+            if parsed is None:
+                continue
+            result.errors.extend(parsed.errors)
+            result.warnings.extend(parsed.warnings)
+            if not parsed.errors:
+                roots.append(parsed.root)
+
+        if roots:
+            result.root = merge_roots(roots, result.errors)
+            resolve_supers(result.root, result.errors)
+
+        run_validations(result.root, self._registry, result.errors, result.warnings)
+        result.root.freeze()
+        return result
+
+    # --- static factories (the 99% case, cross-language consistent) ----
+
+    @classmethod
+    def from_directory(
+        cls,
+        directory: Path | str,
+        providers: list[Provider] | None = None,
+        exclude: list[str] | None = None,
+        recurse: bool = True,
+    ) -> LoadResult:
+        """Load every JSON/YAML file under ``directory`` (recursive by default)."""
+        loader = cls(providers=providers)
+        sources = list(
+            DirectorySource(directory, exclude=exclude, recurse=recurse).expand()
+        )
+        return loader.load(sources)
+
+    @classmethod
+    def from_uris(
+        cls,
+        uris: list[str],
+        providers: list[Provider] | None = None,
+    ) -> LoadResult:
+        """Load metadata from a list of URIs (file:// or http(s)://)."""
+        loader = cls(providers=providers)
+        return loader.load([UriSource(u) for u in uris])
+
+    @classmethod
+    def from_string(
+        cls,
+        content: str,
+        format: MetaDataFormat = MetaDataFormat.JSON,
+        providers: list[Provider] | None = None,
+    ) -> LoadResult:
+        """Load metadata from an in-memory string (defaults to JSON)."""
+        loader = cls(providers=providers)
+        return loader.load([InMemoryStringSource(content, format=format)])
+
+    # --- internals -----------------------------------------------------
+
+    def _parse_source(
+        self, src: MetaDataSource, errors: list[MetaError]
+    ) -> ParseResult | None:
+        """Read a source's content and dispatch to the JSON or YAML parser.
+
+        Returns ``None`` if the source could not be read or its syntax was
+        malformed; the caller appends an error and skips this source from
+        the merge set.
+        """
+        try:
+            text = src.read()
+        except OSError as exc:
+            errors.append(MetaError(str(exc), ErrorCode.ERR_UNKNOWN, src.id))
+            return None
+
+        if src.format is MetaDataFormat.YAML:
+            try:
+                return parse_yaml(text, self._registry, source=src.id)
+            except Exception as exc:  # ParseError carries a code
+                code = getattr(exc, "code", ErrorCode.ERR_MALFORMED_YAML)
+                errors.append(MetaError(str(exc), code, src.id))
+                return None
+
+        # Default: JSON.
         try:
-            return parse_yaml(text, registry, source=path.name)
-        except Exception as exc:  # ParseError from parse_yaml carries a code
-            code = getattr(exc, "code", ErrorCode.ERR_MALFORMED_YAML)
-            errors.append(MetaError(str(exc), code, path.name))
+            doc = json.loads(text)
+        except json.JSONDecodeError as exc:
+            errors.append(MetaError(str(exc), ErrorCode.ERR_MALFORMED_JSON, src.id))
             return None
-    # Default: JSON.
-    try:
-        doc = json.loads(text)
-    except json.JSONDecodeError as exc:
-        errors.append(MetaError(str(exc), ErrorCode.ERR_MALFORMED_JSON, path.name))
-        return None
-    return parse_document(doc, registry, source=path.name)
-
-
-def load_directory(input_dir: str, providers: list[Provider] | None = None) -> LoadResult:
-    registry = compose_registry(providers if providers is not None else [core_provider])
-    result = LoadResult(root=MetaRoot(TYPE_METADATA, SUBTYPE_ROOT, ""))
-
-    # Deterministic ordinal-filename order over JSON + YAML / YML.
-    # Overlay merge is order-sensitive (last-writer-wins on attr conflicts), so
-    # the scan must not depend on filesystem readdir order. Matches the TS
-    # FileMetaDataLoader.loadDirectory contract.
-    accepted = _JSON_SUFFIXES + _YAML_SUFFIXES
-    files = sorted(
-        (p for p in Path(input_dir).iterdir() if p.is_file() and p.suffix.lower() in accepted),
-        key=lambda p: p.name,
-    )
-
-    roots: list[MetaData] = []
-    for path in files:
-        parsed = _parse_file(path, registry, result.errors)
-        if parsed is None:
-            continue
-        result.errors.extend(parsed.errors)
-        result.warnings.extend(parsed.warnings)
-        if not parsed.errors:
-            roots.append(parsed.root)
-
-    if roots:
-        result.root = merge_roots(roots, result.errors)
-        resolve_supers(result.root, result.errors)
-
-    run_validations(result.root, registry, result.errors, result.warnings)
-    result.root.freeze()
-    return result
+        return parse_document(doc, self._registry, source=src.id)

server/python/tests/unit/test_loader_class.py

@@ -0,0 +1,83 @@
+"""Unit tests for the MetaDataLoader class + factory methods.
+
+Covers the Tier-1 invariant: loader.load(sources) -> LoadResult with the
+{root, errors, warnings} shape, plus the three class-method factories
+(from_directory / from_uris / from_string) for the 99% case.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from metaobjects.core_types import core_provider
+from metaobjects.loader.meta_data_loader import LoadResult, MetaDataLoader
+from metaobjects.loader.sources import (
+    DirectorySource,
+    InMemoryStringSource,
+    MetaDataFormat,
+)
+
+
+_TINY_JSON = '{"metadata.root":{"package":"x","children":[]}}'
+
+
+def test_load_from_directory_source(tmp_path: Path) -> None:
+    (tmp_path / "meta.tiny.json").write_text(_TINY_JSON, encoding="utf-8")
+    loader = MetaDataLoader(providers=[core_provider])
+    result = loader.load(list(DirectorySource(tmp_path).expand()))
+    assert isinstance(result, LoadResult)
+    assert result.errors == []
+    assert result.root is not None
+
+
+def test_load_returns_empty_root_on_no_sources() -> None:
+    loader = MetaDataLoader()
+    result = loader.load([])
+    assert result.root is not None
+    assert result.errors == []
+
+
+def test_load_from_inline_yaml() -> None:
+    loader = MetaDataLoader(providers=[core_provider])
+    result = loader.load([
+        InMemoryStringSource(
+            "metadata.root:\n  package: x\n  children: []\n",
+            format=MetaDataFormat.YAML,
+        )
+    ])
+    assert result.errors == []
+    assert result.root is not None
+
+
+def test_load_from_inline_json() -> None:
+    loader = MetaDataLoader(providers=[core_provider])
+    result = loader.load([InMemoryStringSource(_TINY_JSON)])
+    assert result.errors == []
+
+
+def test_load_from_inline_malformed_json_reports_error() -> None:
+    loader = MetaDataLoader(providers=[core_provider])
+    result = loader.load([InMemoryStringSource("not json", id="<bad>")])
+    assert any(e.code.name == "ERR_MALFORMED_JSON" for e in result.errors)
+
+
+def test_from_directory_classmethod(tmp_path: Path) -> None:
+    (tmp_path / "meta.tiny.json").write_text(_TINY_JSON, encoding="utf-8")
+    result = MetaDataLoader.from_directory(tmp_path)
+    assert result.errors == []
+
+
+def test_from_directory_defaults_to_core_provider_when_none(tmp_path: Path) -> None:
+    (tmp_path / "meta.tiny.json").write_text(_TINY_JSON, encoding="utf-8")
+    # Pass providers=None explicitly to assert the default is applied.
+    result = MetaDataLoader.from_directory(tmp_path, providers=None)
+    assert result.errors == []
+
+
+def test_from_string_classmethod() -> None:
+    result = MetaDataLoader.from_string(_TINY_JSON, format=MetaDataFormat.JSON)
+    assert result.errors == []
+
+
+def test_from_string_default_format_is_json() -> None:
+    result = MetaDataLoader.from_string(_TINY_JSON)
+    assert result.errors == []