cdm_data_loaders.converters.facades.pyarrow_to_linkml ¶

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

    Subclasses raised by :func:`generate_linkml_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(PyArrowToLinkMLError):
    """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(PyArrowToLinkMLError):
    """Arrow tables and LinkML record classes generated from Arrow structs require at least one named attribute."""

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

class DuplicateFieldNameError(PyArrowToLinkMLError):
    """LinkML class `attributes` is keyed by slot name and cannot represent duplicate sibling field names."""

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

class NullOnlyTypeError(PyArrowToLinkMLError):
    """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(PyArrowToLinkMLError):
    """Fixed-point decimal representations require `0 <= scale <= precision`; negative scales or `scale > precision` have no plain decimal slot representation."""

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

class UnsupportedTemporalIntervalError(PyArrowToLinkMLError):
    """Relative durations and `month_day_nano` intervals have no lossless built-in `linkml:types` scalar range."""

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

class UnsupportedUnionTypeError(PyArrowToLinkMLError):
    """Arrow type-code unions cannot be represented as typed LinkML slots without class/union redesign."""

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

class UnsupportedArrowTypeError(PyArrowToLinkMLError):
    """Specialized physical encodings such as `pa.run_end_encoded` and custom `pa.ExtensionType` subclasses have no general LinkML mapping."""

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

class EmptyFieldNameError(PyArrowToLinkMLError):
    """LinkML `SlotDefinition` identifiers require a non-empty attribute name; blank Arrow field names cannot form valid LinkML slots."""

    code = "E9"
    title = "Blank or empty field name"

class NestedMultivaluedSlotError(PyArrowToLinkMLError):
    """LinkML slots use a flat `multivalued: true` flag on a single `range` and cannot represent multidimensional lists or lists of map-entry lists without intermediate wrapper classes."""

    code = "E10"
    title = "Directly nested list or map inside a list (`list<list<...>>` or `list<map<...>>`)"

class PyArrowToLinkML(BaseModel):
    """Frozen facade converting PyArrow ``pa.Schema`` objects to LinkML ``SchemaDefinition`` models and YAML."""

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

    schema_name: str | None = None
    schema_id: str | None = None

    def convert(self, schema: pa.Schema) -> SchemaDefinition:
        """Convert a PyArrow ``pa.Schema`` into a LinkML ``SchemaDefinition``.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :return: Emitted LinkML ``SchemaDefinition``.
        :rtype: SchemaDefinition
        """
        return generate_linkml_schema(
            schema,
            schema_name=self.schema_name,
            schema_id=self.schema_id,
        )

    def convert_to_yaml(self, schema: pa.Schema) -> str:
        """Convert a PyArrow ``pa.Schema`` into a LinkML YAML string.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :return: Serialized LinkML YAML string.
        :rtype: str
        """
        return yaml_dumper.dumps(self.convert(schema))

    def convert_to_yaml_file(self, schema: pa.Schema, output: str | Path) -> Path:
        """Convert a PyArrow ``pa.Schema`` and write the LinkML YAML schema to ``output``.

        :param schema: Input PyArrow schema.
        :type schema: pa.Schema
        :param output: Destination YAML file path.
        :type output: str | Path
        :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_yaml(schema), encoding="utf-8")
        return out_path