GraphAnalyzer Objects
class GraphAnalyzer()
downstream_nodes
def downstream_nodes(namespace: str, name: str)
Arguments:
namespace (str): name (str):
Returns:
upstream_nodes
def upstream_nodes(namespace: str, name: str)
Arguments:
namespace (str): name (str):
Returns:
test_delete_node
def test_delete_node(namespace: str, name: str)
Arguments:
namespace (str): name (str):
Returns:
traverse_data_type_violations
def traverse_data_type_violations(
node: NodeTypes,
new_type: str,
path: Optional[List] = None
) -> Generator[Tuple[List[NodeTypes], bool], None, None]
Arguments:
node (NodeTypes): new_type (str):
path
Optional[List], optional - (Default value = None)
Returns:
test_data_type_change
def test_data_type_change(
namespace: str, name: str,
new_type: bool) -> List[Tuple[List[NodeTypes], bool]]
Arguments:
namespace (str): name (str): new_type (bool):
Returns:
traverse_unique_violations
def traverse_unique_violations(
node: NodeTypes,
expects_unique: bool,
path: Optional[List] = None
) -> Generator[Tuple[List[NodeTypes], bool], None, None]
Arguments:
node (NodeTypes): expects_unique (bool):
path
Optional[List], optional - (Default value = None)
Returns:
test_unique_violations
def test_unique_violations(
namespace: str, name: str,
expects_unique: bool) -> List[Tuple[List[NodeTypes], bool]]
Arguments:
namespace (str): name (str):
expects_unique
bool - can't evaluate anything in the case of None
Returns:
traverse_null_violations
def traverse_null_violations(
node: NodeTypes,
is_nullable: bool,
path: Optional[List] = None
) -> Generator[Tuple[List[NodeTypes], bool], None, None]
Arguments:
node (NodeTypes): is_nullable (bool):
path
Optional[List], optional - (Default value = None)
Returns:
test_nullable_violations
def test_nullable_violations(
namespace: str, name: str,
is_nullable: bool) -> List[Tuple[List[NodeTypes], bool]]
Arguments:
namespace (str): name (str):
is_nullable
bool - can't evaluate anything in the case of None
Returns:
column_predecessors
def column_predecessors(namespace: str, name: str)
Arguments:
namespace (str): name (str):
Returns:
column_successors
def column_successors(namespace: str, name: str)
Arguments:
namespace (str): name (str):
Returns: