Commit 8296faab authored by Manuel's avatar Manuel

roleStageDiscovery: added script to fetch information from businessLogic and semanticLinking

parent fce27d02
......@@ -2,7 +2,6 @@ import json
from datetime import datetime
from typing import Dict
class Layer:
'''
This class represents a single layer of the Multilayer Graph.
......@@ -17,12 +16,25 @@ class Layer:
def to_serializable_dict(self, for_db=False) -> Dict:
return {
"layer_name": self.layer_name,
"properties": self.properties
"properties": self.properties,
"use_case": self.use_case,
"total_properties": self.total_properties,
}
@staticmethod
def from_business_logic_dict(layer_info: Dict):
layer = Layer()
layer.layer_name = layer_info["name"]
layer.properties = layer_info["cluster_properties"]
layer.total_properties = layer_info["properties"]
layer.use_case = layer_info["use_case"]
return layer
def from_serializable_dict(self, layer_info: Dict, from_db=False):
self.layer_name = layer_info['layer_name']
self.properties = layer_info['properties']
self.use_case = layer_info["use_case"] if "use_case" in layer_info.keys() else None
self.total_properties = layer_info["total_properties"] if "total_properties"in layer_info.keys() else None
def __repr__(self):
return json.dumps(self.to_serializable_dict())
......
from security.token_manager import TokenManager
import network_constants
from db.entities.layer import Layer
from typing import List, Dict
import requests
import json
def _fetch_layers(use_case: str) -> List[Layer]:
jwt = TokenManager.getInstance().getToken()
url = f'https://{network_constants.BUSINESS_LOGIC_HOSTNAME}:{network_constants.BUSINESS_LOGIC_REST_PORT}/api/use-cases/{use_case}/layers'
response = requests.get(
url,
verify = False,
proxies = { "http":None, "https":None },
headers = {"Authorization": f"Bearer {jwt}"}
)
if response.status_code != 200:
raise ConnectionError(f"Could not fetch data from the server, statuscode: {response.status_code}!")
data = json.loads(response.text)
layers = []
for row in data:
layers.append(Layer.from_business_logic_dict(row))
return layers
def _fetch_nodes(use_case: str, layer: Layer) -> List[Dict]:
jwt = TokenManager.getInstance().getToken()
url = f'https://{network_constants.SEMANTIC_LINKING_HOSTNAME}:{network_constants.SEMANTIC_LINKING_REST_PORT}/api/use-cases/{use_case}/layers/{layer.layer_name}/nodes'
response = requests.get(
url,
verify = False,
proxies = { "http":None, "https":None },
headers = {"Authorization": f"Bearer {jwt}"}
)
if response.status_code != 200:
raise ConnectionError(f"Could not fetch data from the server, statuscode: {response.status_code}!")
data = json.loads(response.text)
nodes = []
for row in data:
nodes.append(row)
return nodes
def fetch_nodes_from_semantic_linking(use_case: str):
layers = _fetch_layers(use_case)
print(str(layers))
for layer in layers:
nodes = _fetch_nodes(use_case, layer)
print("nodes for layer: "+str(nodes))
\ No newline at end of file
import sys
import os
modules_path = '../../../modules/'
if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
import json
import processing.fetching.fetching as f
if __name__ == "__main__":
# TODO read use_case from somewhere
f.fetch_nodes_from_semantic_linking("debug")
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment