Skip to content

cdm_data_loaders.converters.core.metadata

Draft-specific JSON Schema keyword selection for field metadata.

ConversionContext dataclass

Draft-specific metadata selection for a single emission.

Source code in src/cdm_data_loaders/converters/core/metadata.py
@dataclass(frozen=True)
class ConversionContext:
    """Draft-specific metadata selection for a single emission."""

    validator_cls: type
    extra_metadata_keywords: frozenset[str] = frozenset()

    @property
    def metadata_keys(self) -> frozenset[str]:
        """Return the active draft's eligible metadata keywords."""
        return metadata_keys_for(self.validator_cls)

    @property
    def allowed_extra_metadata_keywords(self) -> frozenset[str]:
        """Select standard keywords and explicitly requested extension namespaces."""
        return frozenset(
            key for key in self.extra_metadata_keywords if key in self.metadata_keys or key.startswith("x-")
        )

    @property
    def invalid_extra_metadata_keywords(self) -> frozenset[str]:
        """Return ignored structural, identity and unknown keywords."""
        return self.extra_metadata_keywords - self.allowed_extra_metadata_keywords

metadata_keys property

metadata_keys: frozenset[str]

Return the active draft's eligible metadata keywords.

allowed_extra_metadata_keywords property

allowed_extra_metadata_keywords: frozenset[str]

Select standard keywords and explicitly requested extension namespaces.

invalid_extra_metadata_keywords property

invalid_extra_metadata_keywords: frozenset[str]

Return ignored structural, identity and unknown keywords.

get_known_jsonschema_keywords

get_known_jsonschema_keywords(
    validator_cls: type,
) -> set[str]

Combine draft-specific assertions with the Draft-07 annotation vocabulary.

Source code in src/cdm_data_loaders/converters/core/metadata.py
def get_known_jsonschema_keywords(validator_cls: type) -> set[str]:
    """Combine draft-specific assertions with the Draft-07 annotation vocabulary."""
    keywords = set(Draft7Validator.META_SCHEMA["properties"]) | set(validator_cls.VALIDATORS)
    return keywords - REF_AND_IDENTITY_KEYWORDS

metadata_keys_for cached

metadata_keys_for(validator_cls: type) -> frozenset[str]

Find nonstructural field metadata keywords for a validator class.

Source code in src/cdm_data_loaders/converters/core/metadata.py
@cache
def metadata_keys_for(validator_cls: type) -> frozenset[str]:
    """Find nonstructural field metadata keywords for a validator class."""
    return frozenset(get_known_jsonschema_keywords(validator_cls) - STRUCTURAL_OR_COMPOSITIONAL_KEYWORDS)

decode_field_metadata

decode_field_metadata(
    metadata: dict[bytes, bytes] | None,
) -> dict[str, str]

Decode PyArrow byte key/value metadata into string key/value pairs.

Parameters:

Name Type Description Default
metadata dict[bytes, bytes] | None

Optional raw PyArrow field metadata dictionary.

required

Returns:

Type Description
dict[str, str]

Decoded string-to-string metadata dictionary.

Source code in src/cdm_data_loaders/converters/core/metadata.py
def decode_field_metadata(metadata: dict[bytes, bytes] | None) -> dict[str, str]:
    """Decode PyArrow byte key/value metadata into string key/value pairs.

    :param metadata: Optional raw PyArrow field metadata dictionary.
    :type metadata: dict[bytes, bytes] | None
    :return: Decoded string-to-string metadata dictionary.
    :rtype: dict[str, str]
    """
    if not metadata:
        return {}
    return {k.decode("utf-8"): v.decode("utf-8") for k, v in metadata.items()}