API
grai_source_bigquery
adapters

build_grai_metadata

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

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_table

@build_grai_metadata.register
def build_grai_metadata_from_table(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"
                                  ) -> GenericEdgeMetadataV1

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_table

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

Arguments:

current (Table):

  • 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

def build_metadata(obj, version)

Arguments:

obj: version:

Returns:

adapt_to_client

@multimethod
def adapt_to_client(current: Any,
                    desired: Any) -> Union[SourcedNodeV1, SourcedEdgeV1]

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:

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:

Returns:

make_name

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

Arguments:

node1 (ID): node2 (ID):

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:

Returns:

adapt_list_to_client

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

Arguments:

objs: version: source:

Returns:

workspacebase