cdm_data_loaders.converters.core.extensions ¶

class ExtensionError(ConversionError):
    """An extension namespace or payload violates its registered contract."""

@dataclass(frozen=True, slots=True, init=False)
class ExtensionSpec:
    """An exact namespace spelling and its complete validation schema."""

    namespace: str
    _schema: Mapping[str, Value] | bool | None = field(repr=False, compare=False)
    _schema_loader: ExtensionSchemaLoader | None = field(repr=False, compare=False)

    def __init__(
        self,
        namespace: str,
        schema: Mapping[str, object] | bool | ExtensionSchemaLoader,  # noqa: FBT001
    ) -> None:
        """Validate the namespace and eagerly or lazily own a valid Draft 2020-12 schema.

        :param namespace: ``x-``-prefixed extension namespace key
        :type namespace: str
        :param schema: Draft 2020-12 JSON Schema mapping, boolean schema, or zero-arg lazy loader
        :type schema: Mapping[str, object] | bool | ExtensionSchemaLoader
        :raises ExtensionError: if ``namespace`` or an eager ``schema`` is invalid
        """
        if not isinstance(namespace, str) or not namespace.startswith("x-") or namespace == "x-":
            msg = "Extension namespaces must start with 'x-' and have a name"
            raise ExtensionError(msg)
        object.__setattr__(self, "namespace", namespace)
        if callable(schema):
            object.__setattr__(self, "_schema", None)
            object.__setattr__(self, "_schema_loader", schema)
        else:
            owned = _freeze_and_validate_extension_schema(schema)
            object.__setattr__(self, "_schema", owned)
            object.__setattr__(self, "_schema_loader", None)

    @property
    def schema(self) -> Mapping[str, Value] | bool:
        """Return the validated, deep-frozen Draft 2020-12 schema, loading lazily if needed.

        :return: immutable validation schema
        :rtype: Mapping[str, Value] | bool
        """
        if self._schema is None:
            if self._schema_loader is None:
                msg = f"Extension namespace {self.namespace!r} has no schema configured"
                raise ExtensionError(msg)
            owned = _freeze_and_validate_extension_schema(self._schema_loader())
            object.__setattr__(self, "_schema", owned)
        return self._schema

    def __eq__(self, other: object) -> bool:
        """Compare extension specs by namespace and resolved validation schema.

        :param other: object to compare against
        :type other: object
        :return: ``True`` when ``other`` is an ``ExtensionSpec`` with equal namespace and schema
        :rtype: bool
        """
        if self is other:
            return True
        if not isinstance(other, ExtensionSpec):
            return NotImplemented
        return self.namespace == other.namespace and self.schema == other.schema

    def __hash__(self) -> int:
        """Hash extension spec by namespace and resolved validation schema.

        :return: hash value
        :rtype: int
        """
        return hash((self.namespace, self.schema))

@dataclass(frozen=True, slots=True)
class ExtensionRegistry:
    """A closed set of namespace contracts; additions produce a new registry."""

    specs: tuple[ExtensionSpec, ...] = ()

    def __post_init__(self) -> None:
        """Reject duplicate contracts and isolate the declaration sequence."""
        object.__setattr__(self, "specs", tuple(self.specs))
        if any(not isinstance(spec, ExtensionSpec) for spec in self.specs):
            msg = "Registry entries must be ExtensionSpec instances"
            raise TypeError(msg)
        if len({spec.namespace for spec in self.specs}) != len(self.specs):
            msg = "Duplicate extension namespace"
            raise ExtensionError(msg)

    def register(self, spec: ExtensionSpec, *, replace: bool = False) -> "ExtensionRegistry":
        """Add a contract, requiring an explicit flag to replace an existing one."""
        retained = tuple(existing for existing in self.specs if existing.namespace != spec.namespace)
        if len(retained) != len(self.specs) and not replace:
            msg = f"Already registered: {spec.namespace}"
            raise ExtensionError(msg)
        return ExtensionRegistry((*retained, spec))

    def validate(self, values: Mapping[str, object]) -> Mapping[str, Value]:
        """Validate every namespace and return an isolated immutable payload."""
        owned = freeze_mapping(values)
        specs = {spec.namespace: spec for spec in self.specs}
        for namespace, value in owned.items():
            if namespace not in specs:
                msg = f"Unregistered extension namespace: {namespace}"
                raise ExtensionError(msg)
            try:
                Draft202012Validator(_validation_value(specs[namespace].schema)).validate(_validation_value(value))
            except ValidationError as error:
                msg = f"Invalid {namespace} at {list(error.path)}: {error.message}"
                raise ExtensionError(msg) from error
        return owned

    def extend_payload(self, namespace: str, properties: Mapping[str, object]) -> "ExtensionRegistry":
        """Register additional exact payload keys on a closed object namespace."""
        for spec in self.specs:
            if spec.namespace == namespace:
                schema = mutable_value(spec.schema)
                if not isinstance(schema, dict) or schema.get("type") != "object":
                    msg = f"Not an object namespace: {namespace}"
                    raise ExtensionError(msg)
                existing = schema.setdefault("properties", {})
                if existing.keys() & properties.keys():
                    msg = "Payload keys are already registered"
                    raise ExtensionError(msg)
                existing.update(properties)
                return self.register(ExtensionSpec(namespace, schema), replace=True)
        msg = f"Unregistered extension namespace: {namespace}"
        raise ExtensionError(msg)

@dataclass(frozen=True, slots=True)
class Extensions(Mapping[str, Value]):
    """A validated immutable mapping that retains its validation contract."""

    payload: Mapping[str, Value] = field(default_factory=frozendict)
    registry: ExtensionRegistry = field(default=DEFAULT_EXTENSIONS, repr=False, compare=False)

    def __post_init__(self) -> None:
        """Validate normal construction and own all nested payloads."""
        object.__setattr__(self, "payload", self.registry.validate(self.payload))

    def __getitem__(self, key: str) -> Value:
        """Return a namespace payload."""
        return self.payload[key]

    def __iter__(self) -> Iterator[str]:
        """Iterate namespace names in declaration order."""
        return iter(self.payload)

    def __len__(self) -> int:
        """Return the number of namespaces."""
        return len(self.payload)