Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
objects_properties["property"] = str(properties)
obj_properties_classlist = get_property(
obj1["supportedProperty"])
objects_properties["class_property"] = str(
obj_properties_classlist)
print(objects_node.alias + remove_vocab(obj))
obj_alias = str(objects_node.alias + remove_vocab(obj)).lower()
obj_node = Node(
label="id",
alias=obj_alias,
properties=objects_properties)
redis_graph.add_node(obj_node)
print("commiting objects_property", obj_node)
redis_graph.commit()
print("property node", obj_node)
edge = Edge(objects_node, "has" + remove_vocab(obj), obj_node)
redis_graph.add_edge(edge)
print("commiting objects_property edge", obj_node)
redis_graph.commit()
print("property node", obj_node)
connect_nodes(url_node, "has" +
str(objects_node.alias +
remove_vocab(obj)), obj_node)
if obj_properties_classlist:
return objects_property(
obj_node, obj_properties_classlist, url_node)
else:
return None
# print (obj_node,new_list)
def connect_nodes(source_node, predicate, dest_node):
edge = Edge(source_node, predicate, dest_node)
redis_graph.add_edge(edge)
print("add edge commit", predicate)
redis_graph.commit()
print("edge", edge)
alias="collection_endpoint",
properties=collection_endpoint)
redis_graph.add_node(collection_node)
edge = Edge(url_node, "has_collection_endpoint", collection_node)
redis_graph.add_edge(edge)
# redis_graph.commit()
print("collection endpoint node ", collection_node)
endpointCollection(collection_node, url_node)
# redis_graph.commit()
if classes == 1:
classes_node = Node(
label="id",
alias="classes_endpoint",
properties=classes_endpoint)
redis_graph.add_node(classes_node)
edge = Edge(url_node, "has_classes_endpoint", classes_node)
redis_graph.add_edge(edge)
print("classes endpoint node", classes_node)
endpointclasses(classes_node, url_node)
# redis_graph.commit()
def get_supp_oper(new_file, source_node):
predicate = "supportedoperation"
count = 1
for obj in new_file:
idd = source_node.label + "_id1" + str(count)
node_id = remove_special_character(obj["@id"])
count += 1
node_oper = Node(
label=idd,
alias=node_id,
properties={
idd: str(node_id)})
redis_graph.add_node(node_oper)
edge = Edge(source_node, predicate, node_oper)
redis_graph.add_edge(edge)
get_supp_oper_value(obj, node_oper)
def get_type(source_dict, source_node):
predicate = "type"
node_id = remove_special_character(source_dict["@type"])
node = Node(
label="id1_type",
alias=node_id,
properties={
"id1_type": str(
source_dict["@type"])})
redis_graph.add_node(node)
edge = Edge(source_node, predicate, node)
redis_graph.add_edge(edge)
def get_prop(new_dict, source_node):
predicate = "property"
idd = source_node.label + "_" + "property_id"
s = remove_special_character(new_dict["@id"]).lower()
node_id = s
node_prop_id = Node(
label=idd,
alias=node_id,
properties={
str(idd): str(s)})
redis_graph.add_node(node_prop_id)
edge = Edge(source_node, predicate, node_prop_id)
redis_graph.add_edge(edge)
for obj in new_dict:
if obj != "@id":
if not isinstance(new_dict[obj], list) and not isinstance(
new_dict[obj], dict):
new_obj = remove_special_character(obj)
idd = source_node.label + "_" + new_obj
if new_dict[obj] and not isinstance(new_dict[obj], type(True)):
s = remove_special_character(new_dict[obj]).lower()
else:
s = new_dict[obj]
node_id = remove_special_character(s)
node = Node(
label=idd,
alias=node_id,
properties={
label="id",
alias="collection",
properties=node.properties)
redis_graph.add_node(collection_node)
edge = Edge(node, "get_method", collection_node)
redis_graph.add_edge(edge)
# get_function(node) which shows the properties of node
elif new_obj_op["method"] == "POST":
post_endpoint = 1
return_class = new_obj_op["returns"]
class_node = Node(
label="id",
alias="classes",
properties=node.properties)
redis_graph.add_node(class_node)
edge = Edge(node, "post_method", class_node)
redis_graph.add_edge(edge)
# here we'll call a function for post the endpoint
elif new_obj_op["method"] == "PUT":
put_endpoint = 1
return_class = new_obj_op["returns"]
class_node = Node(
label="id",
alias="classes",
properties=node.properties)
redis_graph.add_node(class_node)
edge = Edge(node, "put_method", class_node)
redis_graph.add_edge(edge)
if post_endpoint == 0 and put_endpoint == 0:
flag = 0
for obj in apidoc["supportedClass"]:
if obj["@id"] == returns: