cdm_data_loaders.coercion.pyarrow ¶

def pyarrow_to_python_type_label(data_type: pa.DataType) -> str:
    """Return the Markdown-formatted Python runtime type label corresponding to a coerced PyArrow ``DataType``.

    Mirrors the scalar and list coercion rules enforced by :func:`coerce_record_for_pyarrow`.

    :param data_type: Target PyArrow data type.
    :type data_type: pa.DataType
    :return: Backtick-wrapped Python runtime type label (e.g. ``"`int`"``, ``"`datetime.datetime` (UTC)"``).
    :rtype: str
    """
    if isinstance(data_type, pa.DictionaryType):
        return pyarrow_to_python_type_label(data_type.value_type)
    if isinstance(
        data_type,
        pa.ListType | pa.LargeListType | pa.ListViewType | pa.LargeListViewType | pa.FixedSizeListType,
    ):
        inner = pyarrow_to_python_type_label(data_type.value_field.type).strip("`")
        return f"`list[{inner}]`"
    if isinstance(data_type, pa.TimestampType):
        tz_suffix = f"({data_type.tz})" if data_type.tz is not None else "(naive)"
        return f"`datetime.datetime` {tz_suffix}"
    for predicate, label in (
        (pa.types.is_boolean, "`bool`"),
        (pa.types.is_integer, "`int`"),
        (pa.types.is_floating, "`float`"),
        (pa.types.is_decimal, "`decimal.Decimal`"),
        (pa.types.is_date, "`datetime.date`"),
        (pa.types.is_time, "`datetime.time`"),
        (_is_any_binary_type, "`bytes`"),
        (_is_any_string_type, "`str`"),
    ):
        if predicate(data_type):
            return label
    return f"`{data_type}`"

def coerce_record_for_pyarrow(
    record: dict[str, Any],
    schema: pa.Schema,
) -> dict[str, Any]:
    """Coerce a JSON-validated record dictionary to match a PyArrow schema.

    Only fields declared in ``schema`` are retained and coerced recursively so
    ``pyarrow.Table.from_pylist`` can construct a table matching ``schema``.

    :param record: JSON-parsed and schema-validated record dictionary.
    :type record: dict[str, Any]
    :param schema: Target PyArrow schema for the record's table.
    :type schema: pa.Schema
    :return: Coerced record dictionary ready for ``DictBuffer`` or ``pa.Table.from_pylist``.
    :rtype: dict[str, Any]
    """
    return {
        field.name: _coerce_value_for_pyarrow(record[field.name], field.type, field.metadata)
        for field in schema
        if field.name in record
    }

def serialize_table_for_jsonschema(table: pa.Table) -> list[dict[str, Any]]:
    """Serialize a ``pa.Table`` into JSON-compatible row dictionaries matching ``generate_json_schema``.

    Encodes ``Decimal`` values as plain fixed-point strings, binary columns as ASCII Base64 strings,
    temporal columns as ISO-8601 strings, and ``pa.map_`` columns as ``[{"key": ..., "value": ...}]`` arrays.

    :param table: Input PyArrow table.
    :type table: pa.Table
    :return: List of JSON-compatible row dictionaries ready for JSON Schema validation.
    :rtype: list[dict[str, Any]]
    """
    return _serialize_table(table, for_linkml=False)

def serialize_record_for_jsonschema(record: Mapping[str, Any], schema: pa.Schema) -> dict[str, Any]:
    """Serialize a single Python row mapping matching ``schema`` for JSON Schema validation.

    :param record: Input Python row dictionary matching ``schema``.
    :type record: Mapping[str, Any]
    :param schema: PyArrow schema describing ``record``.
    :type schema: pa.Schema
    :return: JSON-compatible dictionary matching ``generate_json_schema(schema)``.
    :rtype: dict[str, Any]
    """
    return serialize_table_for_jsonschema(_build_single_row_table(record, schema))[0]

def serialize_table_for_linkml(table: pa.Table) -> list[dict[str, Any]]:
    """Serialize a ``pa.Table`` into JSON-compatible row dictionaries matching ``generate_linkml_schema``.

    Encodes ``Decimal`` values as plain fixed-point strings, binary columns as ASCII Base64 strings,
    temporal columns as ISO-8601 strings, and ``pa.map_`` columns as ``[{"key": ..., "value": ...}]`` entry
    lists while rejecting null list items and nested multivalued list/map items unsupported by LinkML.

    :param table: Input PyArrow table.
    :type table: pa.Table
    :return: List of JSON-compatible row dictionaries ready for LinkML validation.
    :rtype: list[dict[str, Any]]
    """
    return _serialize_table(table, for_linkml=True)

def serialize_record_for_linkml(record: Mapping[str, Any], schema: pa.Schema) -> dict[str, Any]:
    """Serialize a single Python row mapping matching ``schema`` for LinkML validation.

    :param record: Input Python row dictionary matching ``schema``.
    :type record: Mapping[str, Any]
    :param schema: PyArrow schema describing ``record``.
    :type schema: pa.Schema
    :return: JSON-compatible dictionary matching ``generate_linkml_schema(schema)``.
    :rtype: dict[str, Any]
    """
    return serialize_table_for_linkml(_build_single_row_table(record, schema))[0]