"""Manifest parsing and fail-fast validation (ingest-spec §3, §4). Validation is schema-only and runs BEFORE any source call: source existence, credentials, and network access are connector concerns. Queries are configuration, not code — nothing here evaluates manifest content. """ from __future__ import annotations import json import re import urllib.parse from dataclasses import dataclass from pathlib import Path from typing import Any, Union from .errors import ManifestError from .profiles import DEFAULT, BundleProfile _ID_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*\Z") @dataclass(frozen=True) class FileSource: id: str root: str @dataclass(frozen=True) class SqlSource: id: str connection_ref: str @dataclass(frozen=True) class HttpSource: id: str base_url: str credential_ref: Union[str, None] = None Source = Union[FileSource, SqlSource, HttpSource] @dataclass(frozen=True) class Extraction: id: str title: str query: str okf_type: str max_rows: int @dataclass(frozen=True) class Manifest: manifest_version: int source: Source bundle_summary: str extractions: tuple[Extraction, ...] def generated_filename(extraction_id: str, *, profile: BundleProfile = DEFAULT) -> str: """The concept filename for an extraction (spec §5). The `ingest-` prefix keeps the namespace disjoint from `index.md` and `promoted-verdict-*` (spec §3) for every id the §4 grammar admits. The prefix and suffix are the profile's, because the ownership scan globs on the same two values: a name built from one profile and scanned for under another is a file the library cannot recognise as its own. """ return f"{profile.paths.ingest_prefix}{extraction_id}{profile.paths.concept_suffix}" def load_manifest(path: Path) -> Manifest: """Read and fail-fast validate a manifest file (spec §4). Raises ManifestError on any schema violation, before any source call. """ try: raw = path.read_bytes() except OSError as exc: raise ManifestError( f"cannot read manifest {path}: {exc}", code="manifest_unreadable" ) from exc return load_manifest_bytes(raw) def load_manifest_bytes(raw: bytes) -> Manifest: """Fail-fast validate a manifest from its raw bytes (spec §4). The bytes-level entry point exists so materialization can hash and parse the SAME bytes — the §5 provenance stamp must reference exactly the manifest version that produced the run. """ try: data: object = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, ValueError) as exc: raise ManifestError( f"manifest is not valid UTF-8 JSON: {exc}", code="manifest_invalid_json" ) from exc return _validate_manifest(data) def _validate_manifest(data: object) -> Manifest: obj = _require_object(data, "manifest") _require_keys(obj, "manifest", {"manifest_version", "source", "bundle_summary", "extractions"}) version = obj["manifest_version"] if not _is_int(version) or version != 1: raise ManifestError( f"manifest_version must be the integer 1, got {version!r}", code="manifest_version_unsupported", ) source = _validate_source(obj["source"]) bundle_summary = _require_str(obj["bundle_summary"], "bundle_summary", allow_empty=True) extractions = _validate_extractions(obj["extractions"]) return Manifest( manifest_version=1, source=source, bundle_summary=bundle_summary, extractions=extractions, ) def _validate_source(data: object) -> Source: obj = _require_object(data, "source") source_type = obj.get("type") if source_type == "file": _require_keys(obj, "source", {"type", "id", "root"}) return FileSource( id=_validate_id(obj["id"], "source.id"), root=_require_str(obj["root"], "source.root") ) if source_type == "sql": _require_keys(obj, "source", {"type", "id", "connection_ref"}) return SqlSource( id=_validate_id(obj["id"], "source.id"), connection_ref=_require_str(obj["connection_ref"], "source.connection_ref"), ) if source_type == "http": _require_keys(obj, "source", {"type", "id", "base_url"}, optional={"credential_ref"}) credential_ref = obj.get("credential_ref") if credential_ref is not None: credential_ref = _require_str(credential_ref, "source.credential_ref") return HttpSource( id=_validate_id(obj["id"], "source.id"), base_url=_validate_base_url(obj["base_url"]), credential_ref=credential_ref, ) raise ManifestError( f"source.type must be one of 'file', 'sql', 'http', got {source_type!r}", code="source_type_unknown", ) def _validate_base_url(value: object) -> str: base_url = _require_str(value, "source.base_url") # Credentials never live in the manifest (spec §4): no userinfo in the URL. if urllib.parse.urlsplit(base_url).username is not None: raise ManifestError( "source.base_url must not embed credentials; use credential_ref", code="credential_embedded", ) return base_url def _validate_extractions(data: object) -> tuple[Extraction, ...]: if not isinstance(data, list) or not data: raise ManifestError("extractions must be a non-empty list", code="manifest_schema") extractions = tuple(_validate_extraction(entry, index) for index, entry in enumerate(data)) seen: set[str] = set() for extraction in extractions: if extraction.id in seen: raise ManifestError( f"duplicate extraction id {extraction.id!r}", code="extraction_id_duplicate" ) seen.add(extraction.id) return extractions def _validate_extraction(data: object, index: int) -> Extraction: label = f"extractions[{index}]" obj = _require_object(data, label) _require_keys(obj, label, {"id", "title", "query", "okf_type", "max_rows"}) title = _require_str(obj["title"], f"{label}.title") if "\n" in title or "\r" in title: raise ManifestError(f"{label}.title must be single-line", code="manifest_schema") # `[`/`]` would break index-link and navigation parsing (ingest spec §4); the # title is rendered verbatim into the index link and frontmatter (§5/§6), so the # invariant is met by fail-fast validation here, never by downstream repair. if "[" in title or "]" in title: raise ManifestError(f"{label}.title must not contain '[' or ']'", code="manifest_schema") okf_type = _require_str(obj["okf_type"], f"{label}.okf_type") # The verdict layer is RESERVED (spec §3): the promotion gate is the only # path into it — enforced here, fail-fast, before any source call. The # profile decides which types a bundle admits; the door raises its own # typed error, since Door B refuses the same type as a MaterializationError. rejection = DEFAULT.types.rejection(okf_type) if rejection is not None: raise ManifestError(f"{label}.okf_type {rejection.reason}", code=rejection.code) max_rows = obj["max_rows"] if not _is_int(max_rows) or max_rows < 1: raise ManifestError( f"{label}.max_rows must be a positive integer, got {max_rows!r}", code="manifest_schema", ) return Extraction( id=_validate_id(obj["id"], f"{label}.id"), title=title, query=_require_str(obj["query"], f"{label}.query"), okf_type=okf_type, max_rows=max_rows, ) def _validate_id(value: object, label: str) -> str: identifier = _require_str(value, label) if not _ID_PATTERN.match(identifier): raise ManifestError( f"{label} must match [a-z0-9][a-z0-9-]*, got {identifier!r}", code="manifest_schema" ) return identifier def _require_object(data: object, label: str) -> dict[str, Any]: if not isinstance(data, dict): raise ManifestError( f"{label} must be a JSON object, got {type(data).__name__}", code="manifest_schema" ) return data def _require_keys( obj: dict[str, Any], label: str, required: set[str], optional: set[str] | None = None ) -> None: missing = required - obj.keys() if missing: raise ManifestError( f"{label} is missing required field(s): {', '.join(sorted(missing))}", code="manifest_schema", ) unknown = obj.keys() - required - (optional or set()) if unknown: raise ManifestError( f"{label} has unknown field(s): {', '.join(sorted(unknown))}", code="manifest_schema" ) def _require_str(value: object, label: str, *, allow_empty: bool = False) -> str: if not isinstance(value, str): raise ManifestError( f"{label} must be a string, got {type(value).__name__}", code="manifest_schema" ) if not value and not allow_empty: raise ManifestError(f"{label} must be a non-empty string", code="manifest_schema") return value def _is_int(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool)