cdm_data_loaders.converters.facades.pyarrow_to_jsonschema ¶

class PyArrowToJSONSchemaError(ConversionError):
    """Raised when a PyArrow schema, field, or data type cannot be converted to JSON Schema.

    Subclasses raised by :func:`generate_json_schema` carry a stable specification ``code`` and a short
    ``title``; their docstrings state why the construct is rejected. The converter specification document is
    generated from these attributes.
    """

    code: ClassVar[str] = ""
    title: ClassVar[str] = ""

class InvalidPyArrowSchemaError(PyArrowToJSONSchemaError):
    """Schema generation requires a concrete PyArrow `pa.Schema` object; bare `pa.DataType`, `pa.Field`, or Python mappings are not table schemas."""

    code = "E1"
    title = "Input is not a `pa.Schema` instance"

class EmptyStructError(PyArrowToJSONSchemaError):
    """Arrow tables and Parquet struct columns require at least one named child field."""

    code = "E2"
    title = "Empty root schema or nested `pa.struct([])` (zero fields)"

class DuplicateFieldNameError(PyArrowToJSONSchemaError):
    """JSON Schema `properties` is an object mapping keyed by property name and cannot represent duplicate sibling field names."""

    code = "E3"
    title = "Duplicate field name in schema or `pa.struct`"

class NullOnlyTypeError(PyArrowToJSONSchemaError):
    """A column or container element typed solely as `pa.null()` carries no value domain; nullable columns must wrap a concrete Arrow value type with `nullable=True`."""

    code = "E4"
    title = "Standalone `pa.null()` field or element type"

class InvalidDecimalTypeError(PyArrowToJSONSchemaError):
    """Fixed-point decimal string patterns require `0 <= scale <= precision`; negative scales or `scale > precision` have no plain fixed-point string regex."""

    code = "E5"
    title = "Decimal type with negative scale or `scale > precision`"

class UnsupportedTemporalIntervalError(PyArrowToJSONSchemaError):
    """Relative durations and `month_day_nano` intervals have no lossless canonical scalar representation across JSON Schema validators without losing unit or sign-per-component semantics."""

    code = "E6"
    title = "Duration or calendar interval type (`pa.duration`, `pa.month_day_nano_interval`)"

class UnsupportedUnionTypeError(PyArrowToJSONSchemaError):
    """Arrow type-code unions lack stable JSON property discriminators and Parquet storage support; union columns must be normalized before schema export."""

    code = "E7"
    title = "Arrow union type (`pa.sparse_union` or `pa.dense_union`)"

class UnsupportedArrowTypeError(PyArrowToJSONSchemaError):
    """Specialized physical encodings such as `pa.run_end_encoded` and custom `pa.ExtensionType` subclasses have no general target-independent JSON Schema contract."""

    code = "E8"
    title = "Unsupported Arrow physical or extension type (`run_end_encoded`, `ExtensionType`, etc.)"

class PyArrowToJSONSchema(BaseModel):
    """Frozen facade converting PyArrow ``pa.Schema`` objects to Draft 2020-12 JSON Schema documents."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    schema_name: str | None = None
    schema_id: str | None = None
    include_pyarrow_metadata: bool = True
    extension_registry: InstanceOf[ExtensionRegistry] = DEFAULT_EXTENSIONS

    def _emit_schema(
        self,
        schema: pa.Schema,
        *,
        destination: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Read ``schema`` and emit either a standalone JSON Schema or a ``$defs`` entry in ``destination``.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :param destination: Optional namespace JSON Schema dictionary to populate.
        :type destination: dict[str, Any] | None
        :return: Rendered JSON Schema dictionary, or ``destination`` when provided.
        :rtype: dict[str, Any]
        :raises PyArrowToJSONSchemaError: If ``schema`` is invalid or extension validation fails.
        """
        try:
            doc = _reader(
                include_pyarrow_metadata=self.include_pyarrow_metadata,
                extension_registry=self.extension_registry,
            ).read(schema, schema_name=self.schema_name, schema_id=self.schema_id)
            emitter = JsonSchemaEmitter()
            return emitter.emit(doc) if destination is None else emitter.emit_into(doc, destination)
        except PyArrowToJSONSchemaError:
            raise
        except (ExtensionError, ConversionError) as exc:
            raise PyArrowToJSONSchemaError(str(exc)) from exc

    def convert(self, schema: pa.Schema) -> dict[str, Any]:
        """Convert a PyArrow ``pa.Schema`` into a Draft 2020-12 JSON Schema dictionary.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :return: Rendered JSON Schema dictionary.
        :rtype: dict[str, Any]
        """
        return self._emit_schema(schema)

    def convert_into(self, schema: pa.Schema, destination: dict[str, Any]) -> dict[str, Any]:
        """Convert ``schema`` and insert it under ``destination["$defs"][<name>]``.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :param destination: Namespace JSON Schema dictionary to populate.
        :type destination: dict[str, Any]
        :return: Updated ``destination`` dictionary.
        :rtype: dict[str, Any]
        """
        return self._emit_schema(schema, destination=destination)

    def convert_to_json(self, schema: pa.Schema, *, indent: int = 2) -> str:
        """Convert a PyArrow ``pa.Schema`` into a formatted JSON Schema string.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :param indent: JSON indentation spaces (default ``2``).
        :type indent: int
        :return: Formatted JSON string with trailing newline.
        :rtype: str
        """
        return json.dumps(self.convert(schema), indent=indent) + "\n"

    def convert_to_json_file(self, schema: pa.Schema, output: str | Path, *, indent: int = 2) -> Path:
        """Convert a PyArrow ``pa.Schema`` and write the resulting JSON Schema to ``output``.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :param output: Destination file path.
        :type output: str | Path
        :param indent: JSON indentation spaces (default ``2``).
        :type indent: int
        :return: Resolved output ``Path``.
        :rtype: Path
        """
        out_path = Path(output)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(self.convert_to_json(schema, indent=indent), encoding="utf-8")
        return out_path