outsourced wikidata endpoint as external dependency (submodule)

Author: julianweise

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "wikidata_endpoint"]
+	path = wikidata_endpoint
+	url = https://github.com/mpss2019fn1/wikidata_endpoint.git

__init__.py

@@ -1,4 +1,6 @@
 import queue
+import time
+import logging
 from pathlib import Path
 from resources import constant
 
@@ -51,10 +53,15 @@ def _create_cluster(cluster_id, line):
     def fetch_wikidata_ids(self, mysql_connection):
         db_cursor = mysql_connection.cursor()
         entity_ids_list = ",".join([str(x.wikipedia_page_id) for x in self.entities])
+        start_time = time.perf_counter()
         db_cursor.execute(f"SELECT pp_page, pp_value "
                           f"FROM page_props "
                           f"WHERE pp_propname LIKE 'wikibase_item' "
                           f"AND pp_page IN ({entity_ids_list});")
+        end_time = time.perf_counter()
+        logging.info(f"MySQL query execution took {end_time - start_time} seconds for {len(self.entities)} entities")
+
+        start_time = time.perf_counter()
         mapping = {}
         for record in db_cursor:
             if not record[1]:
@@ -72,3 +79,5 @@ def fetch_wikidata_ids(self, mysql_connection):
                 continue
 
             entity.wikidata_id = mapping[entity.wikipedia_page_id]
+        end_time = time.perf_counter()
+        logging.info(f"Mapping WikiData ids took {end_time - start_time} seconds for {len(self.entities)} entities")

cluster.py

@@ -2,10 +2,11 @@
 from collections import Counter
 import queue
 import logging
-import requests
+from pathlib import Path
 
 from resources import constant
-from sparql_endpoint import SparqlEndpoint
+from wikidata_endpoint import WikidataEndpoint
+from wikidata_endpoint import WikidataEndpointConfiguration
 
 
 class Relation:
@@ -67,7 +68,9 @@ def __init__(self, thread_id, work_queue):
         self._cluster_annotations = {"relations": Counter()}
 
         self._mysql_connection = constant.create_mysql_connection()
-        self._sparql_endpoint = SparqlEndpoint(constant.WIKIDATA_API_URL)
+        self._wikidata_endpoint_config = WikidataEndpointConfiguration(Path("resources/wikidata_endpoint_config.ini"))
+        self._wikidata_endpoint = WikidataEndpoint(self._wikidata_endpoint_config)
+        self._chunk_size = constant.STANDARD_CHUNK_SIZE
 
     def run(self):
         while self._analyze_cluster():
@@ -85,28 +88,33 @@ def _analyze_cluster(self):
             return False
 
     def _analyze_entities(self, cluster):
-        chunk_size = constant.STANDARD_CHUNK_SIZE
         index = 0
         metrics = RelationMetrics(len(cluster.entities))
 
         while index < len(cluster.entities):
-            chunk = cluster.entities[index:index + chunk_size]
+            chunk = cluster.entities[index:index + self._chunk_size]
             query = constant.named_entity_relations_sparql_query(chunk)
 
-            try:
-                logging.info(
-                    f"Executing SPARQL query for batch [{index},{index + len(chunk)}] on Thread #{self._thread_id}")
-                relations = [Relation.from_wikidata_record(record) for record in self._sparql_endpoint.query(query)]
-                logging.info(
-                    f"Finished executing SPARQL query on Thread #{self._thread_id}")
-                ClusterAnnotator._count_relations(relations, metrics)
-                index += len(chunk)
-
-            except requests.exceptions.Timeout:
-                chunk_size //= 2
-                pass
+            logging.info(
+                f"Executing SPARQL query for batch [{index},{index + len(chunk)}] on Thread #{self._thread_id}")
+            with self._wikidata_endpoint.request() as request:
+                relations = [Relation.from_wikidata_record(record) for record in
+                             request.post(query,
+                                          on_timeout=self._on_timeout_wikidata_endpoint,
+                                          on_error=self._on_error_wikidata_endpoint)]
+                index += self._chunk_size if len(relations) > 0 else 0
+            logging.info(
+                f"Finished executing SPARQL query on Thread #{self._thread_id}")
+            ClusterAnnotator._count_relations(relations, metrics)
+
         ClusterAnnotator._print_relations(cluster, metrics)
 
+    def _on_timeout_wikidata_endpoint(self, request):
+        self._chunk_size = int(self._chunk_size * (3/4))
+
+    def _on_error_wikidata_endpoint(self, request, error):
+        pass
+
     @staticmethod
     def _count_relations(relations, metrics):
         for relation in relations:

cluster_annotator.py

@@ -0,0 +1,5 @@
+[REMOTE]
+url = https://query.wikidata.org/sparql
+
+[LIMITING]
+concurrent_requests = 5