Create virtuosoQueryRest.py

virtuosoQueryRest.py

@@ -0,0 +1,203 @@
+import requests
+from requests.auth import HTTPDigestAuth, HTTPBasicAuth
+import ssl
+import json
+
+
+
+
+
+
+def execute_query(endpoint, query, auth):
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded',
+        'Accept': 'application/sparql-results+json'
+    }
+
+    params = {
+        'query': query
+    }
+
+    response = requests.post(endpoint, headers=headers, params=params, auth=auth)
+
+    if response.status_code != 200:
+        raise Exception(f"Failed to execute query: {response.text}")
+
+    return response.text
+
+
+def execute_query_no_cache(endpoint, query, auth):
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded',
+        'Accept': 'application/sparql-results+json'
+    }
+
+    params = {
+        'query': query
+    }
+
+    response = requests.post(endpoint, headers=headers, params=params, auth=auth)
+
+    if response.status_code != 200:
+        raise Exception(f"Failed to execute query: {response.text}")
+
+    return response.text
+
+
+def sparqlQuery(endpoint, query, usernameVirt, passwordVirt, USE_CACHE=True):
+    """
+    Make a SPARQL query to a Virtuoso SPARQL endpoint.
+
+    Args:
+    - endpoint (str): The URL of the Virtuoso SPARQL endpoint.
+    - query (str): The SPARQL query to execute.
+    - username (str): The username for authentication.
+    - password (str): The password for authentication.
+
+    Returns:
+    - responseText (requests.Response): The responseText from the Virtuoso SPARQL endpoint.
+    """
+
+    # Use SSL context to establish a secure connection
+    ssl_context = ssl.create_default_context()
+
+    # Try HTTP Digest authentication
+    try:
+        auth = HTTPDigestAuth(usernameVirt, passwordVirt)
+        if USE_CACHE:
+            responseText = execute_query(endpoint, query, auth)
+        else:
+            responseText = execute_query_no_cache(endpoint, query, auth)
+    except Exception as e:
+        print(f"HTTP Digest authentication failed: {str(e)}")
+
+        # Fallback to Basic Auth
+        try:
+            auth = HTTPBasicAuth(usernameVirt, passwordVirt)
+            if USE_CACHE:
+                responseText = execute_query(endpoint, query, auth)
+            else:
+                responseText = execute_query_no_cache(endpoint, query, auth)
+        except Exception as e:
+            print(f"Basic Auth failed: {str(e)}")
+            return None
+
+    return responseText
+
+
+#
+# def sparqlQuery(endpoint, query, usernameVirt, passwordVirt):
+#     """
+#     Make a SPARQL query to a Virtuoso SPARQL endpoint.
+#
+#     Args:
+#     - endpoint (str): The URL of the Virtuoso SPARQL endpoint.
+#     - query (str): The SPARQL query to execute.
+#     - username (str): The username for authentication.
+#     - password (str): The password for authentication.
+#
+#     Returns:
+#     - response (requests.Response): The response from the Virtuoso SPARQL endpoint.
+#     """
+#     headers = {
+#         'Content-Type': 'application/x-www-form-urlencoded',
+#         'Accept': 'application/sparql-results+json'
+#     }
+#
+#     params = {
+#         'query': query
+#     }
+#
+#     # Use SSL context to establish a secure connection
+#     ssl_context = ssl.create_default_context()
+#
+#     # Try HTTP Digest authentication
+#     try:
+#         auth = HTTPDigestAuth(usernameVirt, passwordVirt)
+#         response = requests.post(endpoint, headers=headers, params=params, auth=auth)
+#     except Exception as e:
+#         print(f"HTTP Digest authentication failed: {str(e)}")
+#
+#         # Fallback to Basic Auth
+#         try:
+#             auth = HTTPBasicAuth(usernameVirt, passwordVirt)
+#             response = requests.post(endpoint, headers=headers, params=params, auth=auth)
+#         except Exception as e:
+#             print(f"Basic Auth failed: {str(e)}")
+#             return None
+#
+#     if response.status_code != 200:
+#         raise Exception(f"Failed to execute query: {response.text}")
+#
+#     return response
+
+
+if __name__ == '__main__':
+    # Example usage
+    endpoint = 'https://api-vast.jrc.service.ec.europa.eu/sparql'
+
+    VirtuosoUsername = 'dba'
+    VirtuosoPassword = ''
+    Virtuosokey_filename = 'VIRTUOSO-dba.key'
+
+    USE_CACHE = False # True or False
+
+    #############
+
+    #query = 'SELECT * WHERE { ?s ?p ?o } LIMIT 100'
+
+    # word ="acute sinusitis"
+    # query = f"""
+    #         SELECT ?concept ?label (COUNT(?edge) AS ?score)
+    #         WHERE {{
+    #           ?concept skos:prefLabel|rdfs:label|skos:altLabel|obo:hasRelatedSynonym ?label .
+    #           FILTER (LCASE(STR(?label)) = "{word}")
+    #           ?concept ?edge ?o .
+    #         }}
+    #         GROUP BY ?concept ?label
+    #         ORDER BY DESC(?score)
+    #     """
+
+    choices = ['SNOMED', 'LOINC', 'ICD10', 'MESH', 'NCIT']  # restricts the input to these values only
+
+    # Construct the FROM clauses
+    from_clauses = ' '.join([f"FROM <{choice}>" for choice in choices])
+
+    word = "acute sinusitis"
+    # Construct the full SPARQL query
+    query = f"""
+            SELECT ?concept ?label (COUNT(?edge) AS ?score)
+            {from_clauses}
+            WHERE {{
+              ?concept skos:prefLabel|rdfs:label|skos:altLabel|obo:hasRelatedSynonym ?label .
+              FILTER (LCASE(STR(?label)) = "{word.lower()}")
+              ?concept ?edge ?o .
+            }}
+            GROUP BY ?concept ?label
+            ORDER BY DESC(?score)
+        """
+
+    print(query)
+
+
+
+
+    ###############
+
+    if Virtuosokey_filename:
+        fkeyname = Virtuosokey_filename
+        with open(fkeyname) as f:
+            VirtuosoPassword = f.read()
+
+    responseText = sparqlQuery(endpoint, query, VirtuosoUsername, VirtuosoPassword, USE_CACHE)
+
+    # Parse the response as JSON
+
+    results = json.loads(responseText)
+
+    # Print the results
+    if len(results) > 0 and results['results']['bindings']:
+        for result in results['results']['bindings']:
+            print(result)
+    else:
+        print("!!! VIRTUOSO NO RESULTS !!!")