Source code for tooluniverse.restful_tool
from .graphql_tool import GraphQLTool
import requests
import copy
from .tool_registry import register_tool
[docs]
def execute_RESTful_query(endpoint_url, variables=None):
response = requests.get(endpoint_url, params=variables)
try:
result = response.json()
# Check if the response contains errors
if "error" in result:
print("Invalid Query: ", result["error"])
return False
else:
return result
except requests.exceptions.JSONDecodeError:
print("JSONDecodeError: Could not decode the response as JSON")
return False
except requests.exceptions.HTTPError as e:
print(f"HTTP error occurred: {e}")
except Exception as e:
print(f"An error occurred: {e}")
[docs]
@register_tool("RESTfulTool")
class RESTfulTool(GraphQLTool):
[docs]
def run(self, arguments):
arguments = copy.deepcopy(arguments)
return execute_RESTful_query(
endpoint_url=self.endpoint_url, variables=arguments
)
[docs]
@register_tool("Monarch")
class MonarchTool(RESTfulTool):
[docs]
def __init__(self, tool_config):
endpoint_url = (
"https://api.monarchinitiative.org/v3/api" + tool_config["tool_url"]
)
super().__init__(tool_config, endpoint_url)
[docs]
def run(self, arguments):
arguments = copy.deepcopy(arguments)
query_schema_runtime = copy.deepcopy(self.query_schema)
for key in query_schema_runtime:
if key in arguments:
query_schema_runtime[key] = arguments[key]
if "url_key" in query_schema_runtime:
url_key_name = query_schema_runtime["url_key"]
formatted_endpoint_url = self.endpoint_url.format(
url_key=query_schema_runtime[url_key_name]
)
del query_schema_runtime["url_key"]
else:
formatted_endpoint_url = self.endpoint_url
if isinstance(query_schema_runtime, dict):
print(query_schema_runtime)
if "query" in query_schema_runtime:
query_schema_runtime["q"] = query_schema_runtime[
"query"
] # match with the api
response = execute_RESTful_query(
endpoint_url=formatted_endpoint_url, variables=query_schema_runtime
)
if "facet_fields" in response:
del response["facet_fields"]
def remove_empty_values(obj):
if isinstance(obj, dict):
return {
k: remove_empty_values(v)
for k, v in obj.items()
if v not in [0, [], None]
}
elif isinstance(obj, list):
return [remove_empty_values(v) for v in obj if v not in [0, [], None]]
else:
return obj
response = remove_empty_values(response)
return response
[docs]
@register_tool("MonarchDiseasesForMultiplePheno")
class MonarchDiseasesForMultiplePhenoTool(MonarchTool):
[docs]
def run(self, arguments):
arguments = copy.deepcopy(arguments)
query_schema_runtime = copy.deepcopy(self.query_schema)
for key in query_schema_runtime:
if (key != "HPO_ID_list") and (key in arguments):
query_schema_runtime[key] = arguments[key]
all_diseases = []
for HPOID in arguments["HPO_ID_list"]:
each_query_schema_runtime = copy.deepcopy(query_schema_runtime)
each_query_schema_runtime["object"] = HPOID
each_query_schema_runtime["limit"] = 500
each_output = execute_RESTful_query(
endpoint_url=self.endpoint_url, variables=each_query_schema_runtime
)
each_output = each_output["items"]
each_output_names = [disease["subject_label"] for disease in each_output]
all_diseases.append(each_output_names)
intersection = set(all_diseases[0])
for element in all_diseases[1:]:
intersection &= set(element)
intersection = list(intersection)
if query_schema_runtime["limit"] < len(intersection):
intersection = intersection[: query_schema_runtime["limit"]]
return intersection