def _create_bolts()

in graph/builder/heron/builder.py [0:0]

• 36 lines of code
• 3 McCabe index (conditional complexity)

def _create_bolts(graph_client: GremlinClient, topology_id: str,
                  topology_ref: str,
                  physical_plan: Dict[str, Any],
                  logical_plan: Dict[str, Any]) -> None:

    LOG.info("Creating bolt instance vertices")

    physical_bolts: Dict[str, List[str]] = physical_plan["bolts"]

    counter: int = 0

    for bolt_name in logical_plan["bolts"]:
        LOG.debug("Creating vertices for instances of bolt component: %s",
                  bolt_name)
        for instance_name in physical_bolts[bolt_name]:

            instance: Dict[str, Union[str, int]] = \
                tracker.parse_instance_name(instance_name)

            LOG.debug("Creating vertex for bolt instance: %s",
                      instance_name)

            stream_manager_id: str = \
                physical_plan["instances"][instance_name]["stmgrId"]

            bolt: Vertex = (graph_client.graph_traversal
                            .addV("bolt")
                            .property("container", instance["container"])
                            .property("task_id", instance["task_id"])
                            .property("component", bolt_name)
                            .property("stream_manager", stream_manager_id)
                            .property("topology_id", topology_id)
                            .property("topology_ref", topology_ref)
                            .next())

            # Connect the bolt to its container vertex
            (graph_client.graph_traversal.V(bolt).addE("is_within")
             .to(graph_client.graph_traversal.V()
                 .hasLabel("container")
                 .has("topology_id", topology_id)
                 .has("topology_ref", topology_ref)
                 .has("id", instance["container"])
                 )
             .next())

            counter += 1

    LOG.info("Created %d bolt instances", counter)