data_generator.py 文件源码

def _get_upstream_data(self, dag, node):
        data = {}
        for source, _, edge_attr in dag.in_edges_iter(node, data=True):
            source_attr = dag.node[source]
            source_handler = self._handlers[source_attr['handler']]
            formatted_key_data = source_handler.get(edge_attr['keys'])
            # change the key to template
            data.update({template_key: formatted_key_data[key]
                         for template_key, key in six.viewitems(
                             edge_attr['template_keys'])})
        return data