def load_into_graph()

in hugegraph-llm/src/hugegraph_llm/operators/hugegraph_op/commit_to_hugegraph.py [0:0]

• 53 lines of code
• 22 McCabe index (conditional complexity)

    def load_into_graph(self, vertices, edges, schema):  # pylint: disable=too-many-statements
        # pylint: disable=R0912 (too-many-branches)
        vertex_label_map = {v_label["name"]: v_label for v_label in schema["vertexlabels"]}
        edge_label_map = {e_label["name"]: e_label for e_label in schema["edgelabels"]}
        property_label_map = {p_label["name"]: p_label for p_label in schema["propertykeys"]}

        for vertex in vertices:
            input_label = vertex["label"]
            # 1. ensure the input_label in the graph schema
            if input_label not in vertex_label_map:
                log.critical("(Input) VertexLabel %s not found in schema, skip & need check it!", input_label)
                continue

            input_properties = vertex["properties"]
            vertex_label = vertex_label_map[input_label]
            primary_keys = vertex_label["primary_keys"]
            nullable_keys = vertex_label.get("nullable_keys", [])
            non_null_keys = [key for key in vertex_label["properties"] if key not in nullable_keys]

            has_problem = False
            # 2. Handle primary-keys mode vertex
            for pk in primary_keys:
                if not input_properties.get(pk):
                    if len(primary_keys) == 1:
                        log.error("Primary-key '%s' missing in vertex %s, skip it & need check it again", pk, vertex)
                        has_problem = True
                        break
                    # TODO: transform to Enum first (better in earlier step)
                    data_type = property_label_map[pk]["data_type"]
                    cardinality = property_label_map[pk]["cardinality"]
                    if cardinality == PropertyCardinality.SINGLE.value:
                        input_properties[pk] = default_value_map(data_type)
                    else:
                        input_properties[pk] = []
                    log.warning("Primary-key '%s' missing in vertex %s, mark empty & need check it again!", pk, vertex)
            if has_problem:
                continue

            # 3. Ensure all non-nullable props are set
            for key in non_null_keys:
                if key not in input_properties:
                    self._set_default_property(key, input_properties, property_label_map)

            # 4. Check all data type value is right
            for key, value in input_properties.items():
                # TODO: transform to Enum first (better in earlier step)
                data_type = property_label_map[key]["data_type"]
                cardinality = property_label_map[key]["cardinality"]
                if not self._check_property_data_type(data_type, cardinality, value):
                    log.error("Property type/format '%s' is not correct, skip it & need check it again", key)
                    has_problem = True
                    break
            if has_problem:
                continue

            # TODO: we could try batch add vertices first, setback to single-mode if failed
            vid = self._handle_graph_creation(self.client.graph().addVertex, input_label, input_properties).id
            vertex["id"] = vid

        for edge in edges:
            start = edge["outV"]
            end = edge["inV"]
            label = edge["label"]
            properties = edge["properties"]

            if label not in edge_label_map:
                log.critical("(Input) EdgeLabel %s not found in schema, skip & need check it!", label)
                continue

            # TODO: we could try batch add edges first, setback to single-mode if failed
            self._handle_graph_creation(self.client.graph().addEdge, label, start, end, properties)