Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_source_node method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

import_provenance.py

Source:import_provenance.py

...3from collections import defaultdict4import json5# EXTRACT6################################################################################7def get_source_node(node):8    schannel = Channel.objects.get(id=node.source_channel_id)9    if schannel:10        try:11            snode = schannel.main_tree.get_descendants().get(node_id=node.source_node_id)12            return snode13        except Exception as e:14            print('WARNING: could not find snode for node with studio_id =', node.id, ' node_id =', node.node_id)15            return None16    else:17        return None18def extract_nodes_and_edges(channel_id):19    nodes = {}  # lookup table for ContentNode objects by channel_id, by method, by source_id20    edges = [] # list of tuples (snode.id, node.id) decribing relations "node is imported from snode"21    22    def add_node(node, channel_id, method=None):23        """24        Add the ContentNode object to `nodes` data under `channel_id`, under `method`,25        where `method` is either "added" or "imported".26        """27        if channel_id not in nodes:28            nodes[channel_id] = {29                "added": {},30                "imported": {},31            }32        group_dict = nodes[channel_id][method]33        group_dict[node.id] = node34    35    def extract_imports(channel_id):36        """37        Write the graph infomation to `nodes` and `edges` for `channel_id`.38        """39        channel = Channel.objects.get(id=channel_id)40        node_list = channel.main_tree.get_descendants()41        source_channel_ids = set()42        for node in node_list:43            # split nodes in the according to the medhod of incluion44            # ADDED45            if node.node_id == node.source_node_id:46                add_node(node, channel_id, method='added')47            # IMPORTED48            else:49                # record import relationship as edge tuple50                snode = get_source_node(node)51                if snode:52                    add_node(node, channel_id, method='imported')  # moved here to avoid trouble53                    # print_node_info(snode)54                    edge_tuple = (snode.id, node.id)55                    edges.append(edge_tuple)56                    # queue up source channel_id for recusive call...57                    source_channel_id = snode.get_channel().id58                    assert source_channel_id == node.source_channel_id, 'source_channel_id mismatch!'59                    source_channel_ids.add(source_channel_id)60        61        # recusevily continue to build import-graph data if not already done...62        for source_channel_id in source_channel_ids:63            if source_channel_id not in nodes:64                print('Recursing into channel', source_channel_id)65                extract_imports(source_channel_id)66            else:67                print('No need to extract channel_id', channel_id, 'since already extracted.')68    #69    #70    extract_imports(channel_id)71    return nodes, edges72# interface \/ \/  (nodes, edges )  \/  \/ 73# TRANSFORM74################################################################################75def is_source(edges, studio_id):76    verdict = False77    for edge in edges:78        if edge[0] == studio_id:79            verdict = True80    return verdict81def is_target(edges, studio_id):82    verdict = False83    for edge in edges:84        if edge[1] == studio_id:85            verdict = True86    return verdict87def get_resource_counts_by_kind(subtree):88    node_list = subtree.get_descendants()89    # all_kinds = set([node.kind_id for node in node_list])90    counts = defaultdict(int)91    for node in node_list:92        counts[node.kind_id] += 193    return counts94def get_channel_as_graph_node(channel, name=None, description=None):95    return {96        "channel_id": channel.id,97        "name": channel.name if name is None else name,98        "description": channel.description if description is None else description,99    }100def group_node_list_by_source_channel_id(node_list):101    results = defaultdict(list)102    for node in node_list:103        results[node.source_channel_id].append(node)104    return results105def get_graph(cuv_channel_id, nodes, edges):106    """107    Aggregate the individual edges to granularity of channel_id to form the data108    for the d3 content import vizualization.109    """110    cuv_channel = Channel.objects.get(id=cuv_channel_id)111    graph = {112        "title": "Import graph data for channel " + cuv_channel.name,113        "description": "The channel description of CUV is: " + cuv_channel.description,114        "nodes": {},  # dict {channel_id --> channel_info}, where channe_info is a dict with keys: name, channel_id, counts115        "edges": [],  # edges of the form (source, target, kind, count) where source and target are channel_ids116    }117    # for each channel, there are three graph nodes:118    #  - channel_id119    #  - channel_id+'-added' = to represent nodes added to studio by uploading new content120    #  - channel_id+'-unused' = nodes in a channel that are not imported into derivative channels121    for channel_id, channel_data in nodes.items():122        print('processing channel channel_id='+channel_id)123        channel = Channel.objects.get(id=channel_id)        124        # INPUTS: LISTS OF INDIVIDUAL CONTENT NODES125        added = channel_data["added"].values()126        imported = channel_data["imported"].values()127        all_nodes = added + imported128        # A. ADD GRAPH NODES129        # add three (3x) nodes that correspond to this channel_id130        ########################################################################131        # self132        channel_node = get_channel_as_graph_node(channel)133        channel_node['counts'] = get_resource_counts_by_kind(channel.main_tree)134        graph['nodes'][channel_id] = channel_node135        # added136        added_node_id = channel_id+'-added'137        added_node = get_channel_as_graph_node(138            channel,139            name='Added',140            description='Count of nodes uploaded to channel_id ' + channel_id141        )142        graph['nodes'][added_node_id] = added_node143        # unused144        unused_node_id = channel_id+'-unused'145        unused_node = get_channel_as_graph_node(146            channel,147            name='Unused',148            description='Connts of nodes in channel_id ' + channel_id + ' that are not imported in any downstream channels.'149        )150        graph['nodes'][unused_node_id] = unused_node151        152        153        # B. ADD GRAPH EDGES154        ########################################################################155        # 1. add unused edges156        # counts for the  {{channel_id}}  -->  {{channel_id}}-unused  edges157        unused_aggregates = defaultdict(int)158        for node in all_nodes:159            if not is_source(edges, node.id):160                unused_aggregates[node.kind_id] += 1161        unused_node['counts'] = unused_aggregates162        for kind, count in unused_aggregates.items():163            graph['edges'].append(  (channel_id, unused_node_id, kind, count)  )164        # 165        # 2. add added edges166        # counts for the  {{channel_id}}-added  -->  {{channel_id}}  edges167        added_aggregates = defaultdict(int)168        for node in all_nodes:169            if not is_target(edges, node.id):170                added_aggregates[node.kind_id] += 1171        added_node['counts'] = added_aggregates172        for kind, count in added_aggregates.items():173            graph['edges'].append(  (added_node_id, channel_id, kind, count)   )174        #175        # 3. add imports edges176        # we're computing (snode-->node) imports for current channel only---not recusively177        for source_channel_id, imported_nodes in group_node_list_by_source_channel_id(imported).items():178            print('   processing source_channel_id '+source_channel_id)179            imported_aggregates = defaultdict(int)180            for imported_node in imported_nodes:181                snode = get_source_node(imported_node)182                assert is_source(edges, snode.id), 'failed assumption snode is not in edges'183                imported_aggregates[imported_node.kind_id] += 1184            for kind, count in imported_aggregates.items():185                graph['edges'].append((source_channel_id, channel_id, kind, count))186    #187    # thank you; come again...188    return graph189# TEST CHANNEL190test_channel_id = '8f33b2b268b140fb8bbb6fd09ccc6e17'191nodes, edges = extract_nodes_and_edges(test_channel_id)192g8fe33 = get_graph(test_channel_id, nodes, edges)193print(json.dumps(g8fe33))194# CBSE KA Hindi Class 6 to 9 channel_id= 33c467480fe94f24b4797ef829af9ef6195# contains 1391 nodes, of which  99.93% are imported196# imports  1390 nodes of which:197#   1390 imported from channel Khan Academy (à¤¹à¤¿à¤¨à¥à¤¦à¥, à¤¹à¤¿à¤à¤¦à¥) of which:198#      427 are exercises199#      305 are topics200#      658 are videos201small_channel_id = '33c467480fe94f24b4797ef829af9ef6'202nodes, edges = extract_nodes_and_edges(small_channel_id)203g33c46 = get_graph(small_channel_id, nodes, edges)204print(json.dumps(g33c46))205# SIB Foundation channel_id= 55eea6b34a4c481b8b6adee06a882360206# contains 2141 nodes, of which  99.77% are imported207# imports  2136 nodes of which:208#     10 imported from channel African Storybook of which:209#        10 are html5s210#     8 imported from channel Blockly Games of which:211#        7 are html5s212#        1 are topics213#     928 imported from channel CK-12 of which:214#        334 are exercises215#        8 are html5s216#        56 are topics217#        530 are videos218#     225 imported from channel Foundation 1 of which:219#        225 are videos220#     212 imported from channel Foundation 3 Literacy of which:221#        26 are documents222#        1 are topics223#        185 are videos224#     13 imported from channel Foundation 3 Numeracy of which:225#        2 are topics226#        11 are videos227#     477 imported from channel Nal'ibali Web Resource Tree of which:228#        466 are html5s229#        11 are topics230#     17 imported from channel Pratham Books' StoryWeaver of which:231#        17 are documents232#     76 imported from channel Thoughtful Learning of which:233#        68 are html5s234#        8 are topics235#     170 imported from channel UNKNOWN NAME of which:236#        11 are topics237#        159 are videos238large_channel_id = '55eea6b34a4c481b8b6adee06a882360'239nodes, edges = extract_nodes_and_edges(large_channel_id)240g55ee = get_graph(large_channel_id, nodes, edges)241g55ee['description'] = 'Content imported from AS, Blockly games, Foundatin 1 3 , Pratham, and UKNOWN'242print(json.dumps(g55ee))243# DEBUG CODE244################################################################################245def print_node_info(node, msg=''):246    print(msg)247    print('Node:', 'studio_id='+node.id, 'node_id='+node.node_id, node.title, '('+node.kind_id+')', 'in channel_id='+node.get_channel().id)248    print('      source_channel_id='+node.source_channel_id,249             ',  source_node_id='+node.source_node_id)   250    print('      original_channel_id='+node.original_channel_id,251             ',  original_node_id='+node.original_node_id,252             ',  original_source_node_id='+node.original_source_node_id)253    print('')254# 255# Derivative node:256# Node: studio_id=6e4292e5194a4f4dae3e8e814616a6e7 node_id=ab06de465b8a42aa81ebf4b22c2ca7da Intro to theoretical probability (video) in channel_id=8f33b2b268b140fb8bbb6fd09ccc6e17257#       source_channel_id=d2579f2f62404c32a4791f4311371d0d ,  source_node_id=d0947c31cc4c497b8a99a704a48c79d0258#       original_channel_id=1ceff53605e55bef987d88e0908658c5 ,  original_node_id=02bea4efecbb4baeb6afa4752065ee22 ,  original_source_node_id=655296e4ef4b51efbc57d18028114db5259# 260# >>> snode = get_source_node(dnode)261# >>> print_node_info(snode, 'Source node:')262# Source node:263# Node: studio_id=0617e9e5f1174c47ac7325e67547f1d1 node_id=d0947c31cc4c497b8a99a704a48c79d0 Intro to theoretical probability (video) in channel_id=d2579f2f62404c32a4791f4311371d0d264#       source_channel_id=1ceff53605e55bef987d88e0908658c5 ,  source_node_id=655296e4ef4b51efbc57d18028114db5265#       original_channel_id=1ceff53605e55bef987d88e0908658c5 ,  original_node_id=02bea4efecbb4baeb6afa4752065ee22 ,  original_source_node_id=655296e4ef4b51efbc57d18028114db5266# 267# >>> onode = get_source_node(snode)268# >>> print_node_info(oonode, 'Original node:')269# Original node:270# Node: studio_id=02bea4efecbb4baeb6afa4752065ee22 node_id=655296e4ef4b51efbc57d18028114db5 Intro to theoretical probability (video) in channel_id=1ceff53605e55bef987d88e0908658c5271#       source_channel_id=1ceff53605e55bef987d88e0908658c5 ,  source_node_id=655296e4ef4b51efbc57d18028114db5272#       original_channel_id=1ceff53605e55bef987d88e0908658c5 ,  original_node_id=02bea4efecbb4baeb6afa4752065ee22 ,  original_source_node_id=655296e4ef4b51efbc57d18028114db5273# 274def group_node_list_by_channel_id(node_list):275    results = defaultdict(list)276    for node in node_list:277        channel_id = node.get_channel().id278        results[channel_id].append(node)279    return results280def get_node_by_studio_id(studio_id):281    for node_list in nodes.values():282        if studio_id in node_list:283            return node_list[studio_id]284    return None285provenance_test_channel = Channel.objects.get(id='8f33b2b268b140fb8bbb6fd09ccc6e17')286dtopic = provenance_test_channel.main_tree.children.get(title='Basic theoretical probability')287# dnode = dtopic.children.all()[0]288# 289# print_node_info(dnode, 'Derivative node:')290# snode = get_source_node(dnode)291# print_node_info(snode, 'Source node:')292# onode = get_source_node(snode)...

graph.py

Source:graph.py

...79        self.dest_node = dest_node80        self.travel_time = travel_time81        self.road_type = road_type82        self.traffic_multiplier = traffic_multiplier83    def get_source_node(self):84        """85        Getter method for DirectedRoad86        Returns:87        Node, representing the source node88        """89        return self.src_node90    def get_destination_node(self):91        """92        Getter method for DirectedRoad93        Returns:94        Node, representing the destination node95        """96        return self.dest_node97    def get_road_type(self):98        """99        Getter method for DirectedRoad100        Returns:101        str, representing the road type of the road102        """103        return self.road_type104    def get_travel_time(self, has_traffic=False):105        """106        Gets the travel_time for this road. If there is traffic,107        - multiply the time it takes to travel on a road by its traffic multiplier108        Parameter:109        has_traffic - bool, True if there is traffic, False otherwise110        Returns:111        float, representing the time to travel from the source node to the destination node112        """113        if has_traffic:114            return self.travel_time * self.get_traffic_multiplier()115        return self.travel_time116    def get_traffic_multiplier(self):117        """118        Getter method for DirectedRoad119        Returns:120        float, representing the traffic multiplier121        """122        return self.traffic_multiplier123    def __str__(self):124        """125        Function that is called when print() is called on a DirectedRoad object.126        Returns:127        str, with the format 'src -> dest takes travel_time minute(s) via road_type road with traffic multiplier traffic_multiplier'128        Note: For the total time assume normal traffic conditions129        """130        src_node = self.get_source_node().get_name()131        dest_node = self.get_destination_node().get_name()132        travel_time = str(self.get_travel_time())133        road_type = self.get_road_type()134        traffic_multiplier = str(self.get_traffic_multiplier())135        return src_node + " -> " + dest_node + " takes " + travel_time + \136            " minute(s) via " + road_type + " road with traffic multiplier " + traffic_multiplier137    def __hash__(self):138        return self.__str__().__hash__()139# PROBLEM 1: Implement methods of this class based on the given docstring.140# DO NOT CHANGE THE FUNCTIONS THAT HAVE BEEN IMPLEMENTED FOR YOU.141class RoadMap():142    """Represents a road map -> a directed graph of Node and DirectedRoad objects"""143    def __init__(self):144        """145        Initalizes a new instance of RoadMap.146        """147        self.nodes = set()148        self.nodes_to_roads = {}  # must be a dictionary of Node -> list of roads starting at that node149    def __str__(self):150        """151        Function that is called when print() is called on a RoadMap object.152        Returns:153        str, representation of the RoadMap.154        """155        road_strs = []156        for roads in self.nodes_to_roads.values():157            for road in roads:158                road_strs.append(str(road))159        road_strs = sorted(road_strs)  # sort alphabetically160        return '\n'.join(road_strs)  # concat road_strs with "\n"s between them161    def get_all_nodes(self):162        """163        Return:164        a COPY of all nodes in the RoadMap. Does not modify self.nodes165        """166        return self.nodes.copy()167    def contains_node(self, node):168        """169        Param:170        node Node object171        Return:172        True, if node is in the graph. False, otherwise.173        """174        if node in self.nodes:175            return True176        else:177            return False178    def insert_node(self, node):179        """180        Adds a Node object to the RoadMap.181        Raises a ValueError if it is already in the graph.182        Param:183        node object184        """185        if self.contains_node(node):186            raise ValueError187        else:188            self.nodes.add(node)189            self.nodes_to_roads[node] = []190    def insert_road(self, road):191        """192        Adds a DirectedRoad instance to the RoadMap.193        Raises a ValueError if either of the nodes associated with the road is not in the graph.194        Param:195        road, DirectedRoad object196        """197        if self.contains_node(road.get_source_node()) and self.contains_node(road.get_destination_node()):198            self.nodes_to_roads[road.get_source_node()].append(road)199        else:200            raise ValueError201    def get_reachable_roads_from_node(self, node, restricted_roads):202        """203        Get the roads out of Node node, excluding roads whose types are in restricted_roads204        Param:205            node: Node206            find reachable roads out of this node207            restricted_roads: List of strings (types of roads)208            road types that cannot be traveled on209        Return:210            A new list of all the roads that start at given node,211            whose types are not in restricted_roads.212            Empty list if the node is not in the graph....

Edge.py

Source:Edge.py

...8  def set_source_node(self, source_node):9    self.source_node = source_node10  def set_destination_node(self, destination_node):11    self.destination_node = destination_node12  def get_source_node(self):13    return self.source_node14  def get_destination_node(self):15    return self.destination_node16  def print_edge(self):17    self.get_source_node().print_node()18    self.get_destination_node().print_node()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.