cdm_data_loaders.converters.facades.pyiceberg_to_linkml ¶

def convert_type(
    field_type: IcebergType, *, extension_registry: ExtensionRegistry = DEFAULT_EXTENSIONS
) -> SlotDefinition:
    """Read and emit a single Iceberg type.

    :param field_type: Iceberg type to convert.
    :type field_type: IcebergType
    :param extension_registry: Registry for Iceberg extensions.
    :type extension_registry: ExtensionRegistry
    :returns: LinkML slot definition.
    """
    return LinkMLEmitter().emit_node(IcebergReader(extension_registry, mapping_target="LinkML").read_node(field_type))

def convert_field(field: NestedField, *, extension_registry: ExtensionRegistry = DEFAULT_EXTENSIONS) -> SlotDefinition:
    """Read and emit a field with its cardinality and description.

    :param field: Iceberg field to convert.
    :type field: NestedField
    :param extension_registry: Registry for Iceberg extensions.
    :type extension_registry: ExtensionRegistry
    :returns: Named LinkML slot definition.
    """
    return LinkMLEmitter().emit_field(IcebergReader(extension_registry, mapping_target="LinkML").read_field(field))

def convert_struct(
    fields: tuple[NestedField, ...], *, extension_registry: ExtensionRegistry = DEFAULT_EXTENSIONS
) -> SlotDefinition:
    """Read and emit a struct without a document envelope.

    :param fields: Iceberg struct fields to convert.
    :type fields: tuple[NestedField, ...]
    :param extension_registry: Registry for Iceberg extensions.
    :type extension_registry: ExtensionRegistry
    :returns: LinkML slot referencing the struct class.
    """
    return convert_type(StructType(*fields), extension_registry=extension_registry)

def table_to_linkml(
    table: Table,
    identifier: tuple[str, ...],
    *,
    extension_registry: ExtensionRegistry = DEFAULT_EXTENSIONS,
    schema: SchemaDefinition | None = None,
) -> SchemaDefinition:
    """Read loaded table metadata and emit a LinkML schema model without I/O.

    :param table: Loaded Iceberg table.
    :type table: Table
    :param identifier: Namespace and table name.
    :type identifier: tuple[str, ...]
    :param extension_registry: Registry for Iceberg extensions.
    :type extension_registry: ExtensionRegistry
    :param schema: Optional schema to append table classes to, preserving existing definitions.
    :type schema: SchemaDefinition | None
    :returns: Schema model suitable for LinkML's YAML dumper.
    """
    document = IcebergReader(extension_registry, mapping_target="LinkML").read_table(table, identifier)
    emitter = LinkMLEmitter()
    return emitter.emit(document) if schema is None else emitter.emit_into(document, schema)