def filter_item()

in hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py [0:0]

• 29 lines of code
• 10 McCabe index (conditional complexity)

def filter_item(schema, items) -> List[Dict[str, Any]]:
    # filter vertex and edge with invalid properties
    filtered_items = []
    properties_map = {"vertex": {}, "edge": {}}
    for vertex in schema["vertexlabels"]:
        properties_map["vertex"][vertex["name"]] = {
            "primary_keys": vertex["primary_keys"],
            "nullable_keys": vertex["nullable_keys"],
            "properties": vertex["properties"]
        }
    for edge in schema["edgelabels"]:
        properties_map["edge"][edge["name"]] = {
            "properties": edge["properties"]
        }
    log.info("properties_map: %s", properties_map)
    for item in items:
        item_type = item["type"]
        if item_type == "vertex":
            label = item["label"]
            non_nullable_keys = (
                set(properties_map[item_type][label]["properties"])
                .difference(set(properties_map[item_type][label]["nullable_keys"])))
            for key in non_nullable_keys:
                if key not in item["properties"]:
                    item["properties"][key] = "NULL"
        for key, value in item["properties"].items():
            if not isinstance(value, str):
                item["properties"][key] = str(value)
        filtered_items.append(item)

    return filtered_items