mock-up version

Author: prohde

DeTrusty/Configuration/__init__.py

@@ -0,0 +1,14 @@
+__author__ = "Philipp D. Rohde"
+
+
+def get_datasource_description(endpoints):
+    """At this point in time, only the endpoint URLs are stored.
+    In a later version, a more sophisticated datasource description will be implemented."""
+    endpoints = [e.strip() for e in endpoints.split(',') if e]
+    with open('/DeTrusty/Config/datasource_description.txt', 'w') as f:
+        for endpoint in endpoints:
+            f.write(endpoint + "\n")
+
+
+def parse_datasource_file(datasource_file):
+    return [e.strip() for e in open(datasource_file, 'r').readlines()]

DeTrusty/Configuration/get_datasource_description.py

@@ -0,0 +1,37 @@
+__author__ = "Philipp D. Rohde"
+
+
+class Query:
+    """Simple representation of a SPARQL query. Object to be changed in later versions."""
+
+    def __init__(self, query_string):
+        self.query_string = query_string
+        self.distinct = True if "distinct" in query_string.lower() else False
+
+        variables = query_string.lower().split("where", 1)[0]
+        variables = variables.lower().replace("select", "").replace("distinct", "")
+        variables = ''.join(variables.split())
+        variables = variables.split("?")
+        self.variables = []
+        for variable in variables:
+            if variable:
+                self.variables.append(variable.strip())
+
+        prefixes = query_string.split("SELECT", 1)[0]
+        if prefixes:
+            prefixes = prefixes.split("\n")
+            self.prefixes = [(prefix.strip()[7:prefix.strip().find(':')], prefix.strip()[prefix.strip().find(':')+1:]) for prefix in prefixes if prefix]
+        else:
+            self.prefixes = []
+
+    def get_variables(self):
+        return self.variables
+
+    def get_query_string(self):
+        return self.query_string
+
+    def get_distinct(self):
+        return self.distinct
+
+    def get_prefixes(self):
+        return self.prefixes

DeTrusty/Decomposer/Query.py

@@ -0,0 +1,35 @@
+"""
+Created on Dec 11, 2013
+Implements the Xdistinct operator.
+The intermediate results are represented in a queue.
+@author: Maribel Acosta Deibe
+"""
+from multiprocessing import Queue
+
+
+class Xdistinct(object):
+
+    def __init__(self, vars):
+        # self.input       = Queue()
+        self.qresults = Queue()
+        self.vars = vars
+        self.bag = {}
+
+    def execute(self, left, dummy, out, processqueue=Queue()):
+        # Executes the Xdistinct.
+        self.left = left
+        self.qresults = out
+        tuple = self.left.get(True)
+
+        while (not (tuple == "EOF")):
+            str_tuple = str(sorted(tuple.items()))
+            get = self.bag.get(str_tuple, False)
+
+            if not (get):
+                self.qresults.put(tuple)
+                self.bag.update({str_tuple: True})
+            tuple = self.left.get(True)
+
+        # Put EOF in queue and exit.
+        self.qresults.put("EOF")
+        return

DeTrusty/Decomposer/__init__.py

@@ -0,0 +1,129 @@
+"""
+Created on Jul 10, 2011
+Implements the Xunion operator.
+The intermediate results are represented in a queue.
+@author: Maribel Acosta Deibe
+"""
+from multiprocessing import Queue
+from multiprocessing.queues import Empty
+
+
+class Xunion(object):
+
+    def __init__(self, vars_left, vars_right):
+        self.left = Queue()
+        self.right = Queue()
+        self.qresults = Queue()
+        self.vars_left = vars_left
+        self.vars_right = vars_right
+        self.count = 0
+
+    def instantiate(self, d):
+        newvars_left = self.vars_left - set(d.keys())
+        newvars_right = self.vars_right - set(d.keys())
+        # return Xunion(newvars_left, newvars_right, self.distinct)
+        return Xunion(newvars_left, newvars_right)
+
+    def instantiateFilter(self, instantiated_vars, filter_str):
+        newvars_left = self.vars_left - set(instantiated_vars)
+        newvars_right = self.vars_right - set(instantiated_vars)
+        return Xunion(newvars_left, newvars_right)
+
+    def execute(self, left, right, out, processqueue=Queue()):
+        # Executes the Xunion.
+        self.left = left
+        self.right = right
+        self.qresults = out
+        # print "left", hex(id(left)), "right", hex(id(right)), "out", hex(id(out))
+
+        # Identify the kind of union to perform.
+        if self.vars_left == self.vars_right:
+            self.sameVariables()
+        else:
+            self.differentVariables()
+
+        # Put EOF in queue and exit.
+        self.qresults.put("EOF")
+
+    def sameVariables(self):
+        # Executes the Xunion operator when the variables are the same.
+
+        # Initialize tuples.
+        tuple1 = None
+        tuple2 = None
+
+        # Get the tuples from the queues.
+        while tuple1 != "EOF" or tuple2 != "EOF":
+            if tuple1 != "EOF":
+                try:
+                    tuple1 = self.left.get(False)
+                    if tuple1 != "EOF":
+                        self.count += 1
+                        self.qresults.put(tuple1)
+                except Empty:
+                    # This catch:
+                    # Empty: in tuple1 = self.left.get(False), when the queue is empty.
+                    pass
+
+            if tuple2 != "EOF":
+                try:
+                    tuple2 = self.right.get(False)
+                    if tuple2 != "EOF":
+                        self.count += 1
+                        self.qresults.put(tuple2)
+
+                except Empty:
+                    # This catch:
+                    # Empty: in tuple2 = self.right.get(False), when the queue is empty.
+                    pass
+
+    def differentVariables(self):
+        # Executes the Xunion operator when the variables are not the same.
+
+        # Initialize tuples.
+        tuple1 = None
+        tuple2 = None
+
+        # Initialize empty tuples.
+        v1 = {}
+        v2 = {}
+
+        # Add empty values to variables of the other argument.
+        for v in self.vars_right:
+            v1.update({v: ''})
+
+        for v in self.vars_left:
+            v2.update({v: ''})
+
+        # Get the tuples from the queues.
+        while (not (tuple1 == "EOF") or not (tuple2 == "EOF")):
+
+            # Get tuple from left queue, and concatenate with empty tuple.
+            if not (tuple1 == "EOF"):
+                try:
+                    tuple1 = self.left.get(False)
+                    if not (tuple1 == "EOF"):
+                        res = {}
+                        res.update(v1)
+                        res.update(tuple1)
+                        self.qresults.put(res)
+                        # print(tuple1)
+                except Exception:
+                    # This catch:
+                    # Empty: in tuple1 = self.left.get(False), when the queue is empty.
+                    pass
+
+            # Get tuple from right queue, and concatenate with empty tuple.
+            if not (tuple2 == "EOF"):
+                try:
+                    tuple2 = self.right.get(False)
+                    if not (tuple2 == "EOF"):
+                        res = {}
+                        res.update(v2)
+                        res.update(tuple2)
+                        self.qresults.put(res)
+                        # print(tuple2)
+                except Exception:
+                    # This catch:
+                    # Empty: in tuple2 = self.right.get(False), when the queue is empty.
+                    pass

DeTrusty/Operators/Xdistinct.py

@@ -0,0 +1,121 @@
+import urllib.parse
+import urllib.request
+import logging
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+handler = logging.StreamHandler()
+handler.setLevel(logging.INFO)
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+handler.setFormatter(formatter)
+logger.addHandler(handler)
+
+
+def contact_source(server, query, queue, buffersize=16384, limit=-1):
+    # Contacts the datasource (i.e. real endpoint).
+    # Every tuple in the answer is represented as Python dictionaries
+    # and is stored in a queue.
+    logger.info("Contacting endpoint: " + server)
+    b = None
+    cardinality = 0
+
+    referer = server
+    server = server.split("http://")[1]
+    if '/' in server:
+        (server, path) = server.split("/", 1)
+    else:
+        path = ""
+    host_port = server.split(":")
+    port = 80 if len(host_port) == 1 else host_port[1]
+
+    if limit == -1:
+        b, cardinality = contact_source_aux(referer, server, path, port, query, queue)
+    else:
+        # Contacts the datasource (i.e. real endpoint) incrementally,
+        # retrieving partial result sets combining the SPARQL sequence
+        # modifiers LIMIT and OFFSET.
+
+        # Set up the offset.
+        offset = 0
+
+        while True:
+            query_copy = query + " LIMIT " + str(limit) + " OFFSET " + str(offset)
+            b, card = contact_source_aux(referer, server, path, port, query_copy, queue)
+            cardinality += card
+            if card < limit:
+                break
+
+            offset = offset + limit
+
+    # Close the queue
+    queue.put("EOF")
+    return b
+
+
+def contact_source_aux(referer, server, path, port, query, queue):
+    # Setting variables to return.
+    b = None
+    cardinality = 0
+
+    if '0.0.0.0' in server:
+        server = server.replace('0.0.0.0', 'localhost')
+
+    js = "application/sparql-results+json"
+    params = {'query': query, 'format': js}
+    headers = {"User-Agent":
+                   "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.77 Safari/537.36",
+               "Accept": js}
+    try:
+        data = urllib.parse.urlencode(params)
+        data = data.encode('utf-8')
+        req = urllib.request.Request(referer, data, headers)
+        with urllib.request.urlopen(req) as response:
+            resp = response.read()
+            resp = resp.decode()
+            res = resp.replace("false", "False")
+            res = res.replace("true", "True")
+            res = eval(res)
+            if type(res) == dict:
+                b = res.get('boolean', None)
+
+                if 'results' in res:
+                    # print "raw results from endpoint", res
+                    for x in res['results']['bindings']:
+                        for key, props in x.items():
+                            # Handle typed-literals and language tags
+                            suffix = ''
+                            if props['type'] == 'typed-literal':
+                                if isinstance(props['datatype'], bytes):
+                                    suffix = "^^<" + props['datatype'].decode('utf-8') + ">"
+                                else:
+                                    suffix = "^^<" + props['datatype'] + ">"
+                            elif "xml:lang" in props:
+                                suffix = '@' + props['xml:lang']
+                            try:
+                                if isinstance(props['value'], bytes):
+                                    x[key] = props['value'].decode('utf-8') + suffix
+                                else:
+                                    x[key] = props['value'] + suffix
+                            except:
+                                x[key] = props['value'] + suffix
+
+                        queue.put(x)
+                        cardinality += 1
+                    # Every tuple is added to the queue.
+                    #for elem in reslist:
+                        # print elem
+                        #queue.put(elem)
+
+            else:
+                print("the source " + str(server) + " answered in " + res.getheader("content-type") + " format, instead of"
+                       + " the JSON format required, then that answer will be ignored")
+    except Exception as e:
+        logger.error("Exception while sending request to ", referer, "msg:", e)
+        print("Exception while sending request to ", referer, "msg:", e)
+
+    # print "b - ", b
+    # print server, query, len(reslist)
+
+    # print "Contact Source returned: ", len(reslist), ' results'
+    return b, cardinality