API
grai_source_openlineage
adapters

build_grai_metadata

@multimethod
def build_grai_metadata(current: Any, desired: Any) -> None

Arguments:

current (Any): desired (Any):

Returns:

build_grai_metadata_from_column

@build_grai_metadata.register
def build_grai_metadata_from_column(current: Column,
                                    version: Literal["v1"] = "v1"
                                    ) -> ColumnMetadata

Arguments:

current (Column):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_grai_metadata_from_node

@build_grai_metadata.register
def build_grai_metadata_from_node(current: Table,
                                  version: Literal["v1"] = "v1"
                                  ) -> TableMetadata

Arguments:

current (Table):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_grai_metadata_from_edge

@build_grai_metadata.register
def build_grai_metadata_from_edge(current: Edge,
                                  version: Literal["v1"] = "v1"
                                  ) -> EdgeReturnTypes

Arguments:

current (Edge):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_app_metadata

@multimethod
def build_app_metadata(current: Any, desired: Any) -> None

Arguments:

current (Any): desired (Any):

Returns:

build_metadata_from_column

@build_app_metadata.register
def build_metadata_from_column(current: Column,
                               version: Literal["v1"] = "v1") -> Dict

Arguments:

current (Column):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_metadata_from_edge

@build_app_metadata.register
def build_metadata_from_edge(current: Edge,
                             version: Literal["v1"] = "v1") -> Dict

Arguments:

current (Edge):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_metadata_from_node

@build_app_metadata.register
def build_metadata_from_node(current: Table,
                             version: Literal["v1"] = "v1") -> Dict

Arguments:

current (Table):

  • version Literal["v1"], optional - (Default value = "v1")

Returns:

build_metadata

def build_metadata(obj, version)

Arguments:

obj: version:

Returns:

adapt_to_client

@multimethod
def adapt_to_client(current: Any, desired: Any)

Arguments:

current (Any): desired (Any):

Returns:

adapt_column_to_client

@adapt_to_client.register
def adapt_column_to_client(current: Column, source: SourceSpec,
                           version: Literal["v1"]) -> SourcedNodeV1

Arguments:

current: source:

  • version - (Default value = "v1")

Returns:

adapt_table_to_client

@adapt_to_client.register
def adapt_table_to_client(current: Table, source: SourceSpec,
                          version: Literal["v1"]) -> SourcedNodeV1

Arguments:

current: source:

  • version - (Default value = "v1")

Returns:

make_name

def make_name(node1: NodeTypes, node2: NodeTypes) -> str

Arguments:

node1 (NodeTypes): node2 (NodeTypes):

Returns:

adapt_edge_to_client

@adapt_to_client.register
def adapt_edge_to_client(current: Edge, source: SourceSpec,
                         version: Literal["v1"]) -> SourcedEdgeV1

Arguments:

current: source:

  • version - (Default value = "v1")

Returns:

adapt_seq_to_client

@adapt_to_client.register
def adapt_seq_to_client(
        objs: Sequence, source: SourceSpec,
        version: Literal["v1"]) -> List[Union[SourcedNodeV1, SourcedEdgeV1]]

Arguments:

objs: source: version:

Returns:

adapt_list_to_client

@adapt_to_client.register
def adapt_list_to_client(
        objs: List, source: SourceSpec,
        version: Literal["v1"]) -> List[Union[SourcedNodeV1, SourcedEdgeV1]]

Arguments:

objs: source: version:

Returns:

package_definitionsbase