Wikidata Elastic Search - Backend Implementation

sand/app.py

@@ -37,25 +37,18 @@
         generate_readonly_api_4dict(
             "entities",
             serialize=serialize_entity,
-            id2ent=ChainedMapping(
-                EntityAR(), import_attr(SETTINGS["entity"]["default"])
-            ),
+            id2ent=EntityAR(),
         ),
         generate_readonly_api_4dict(
             "classes",
             serialize=serialize_class,
-            id2ent=ChainedMapping(
-                OntClassAR(), import_attr(SETTINGS["ont_classes"]["default"])
-            ),
+            id2ent=OntClassAR(),
             unique_field_funcs={"uri": OntClass.uri2id},
         ),
         generate_readonly_api_4dict(
             "properties",
             serialize=serialize_property,
-            id2ent=ChainedMapping(
-                OntPropertyAR(),
-                import_attr(SETTINGS["ont_props"]["default"]),
-            ),
+            id2ent=OntPropertyAR(),
             unique_field_funcs={"uri": OntProperty.uri2id},
         ),
     ],

sand/controllers/search.py

@@ -1,18 +1,21 @@
 import threading
-from typing import Dict, List
+from typing import Dict, List, Union
 from flask.blueprints import Blueprint
 from sm.misc.funcs import import_func
 from sand.config import SETTINGS
 from flask import request, jsonify
 
-from sand.extension_interface.search import ISearch
+from sand.extension_interface.search import IEntitySearch, IOntologySearch
+from sand.models.search import SearchPayload
+from gena.serializer import get_dataclass_serializer
 
 search_bp = Blueprint("search", "search")
 
 GetSearchCache = threading.local()
+serializer = get_dataclass_serializer(SearchPayload)
 
 
-def get_search(name) -> ISearch:
+def get_search(name) -> Union[IEntitySearch, IOntologySearch]:
     """
     Returns an implementation of an ISearch Interface from the
     configuration file.
@@ -33,23 +36,26 @@ def search_classes():
     """API Route to search for classes with their names"""
     search_text = request.args.get('q')
     wikidata_search = get_search('classes')
-    payload = wikidata_search.find_class_by_name(search_text)
-    return jsonify(payload)
+    search_payload = wikidata_search.find_class_by_name(search_text)
+    serialized_payload = serializer(search_payload)
+    return jsonify(serialized_payload)
 
 
 @search_bp.route(f"/{search_bp.name}/entities", methods=["GET"])
 def search_entities():
     """API Route to search for entities with their names"""
     search_text = request.args.get('q')
     wikidata_search = get_search('entities')
-    payload = wikidata_search.find_entity_by_name(search_text)
-    return jsonify(payload)
+    search_payload = wikidata_search.find_entity_by_name(search_text)
+    serialized_payload = serializer(search_payload)
+    return jsonify(serialized_payload)
 
 
 @search_bp.route(f"/{search_bp.name}/props", methods=["GET"])
 def search_props():
     """API Route to search for properties with their names"""
     search_text = request.args.get('q')
     wikidata_search = get_search('props')
-    payload = wikidata_search.find_props_by_name(search_text)
-    return jsonify(payload)
+    search_payload = wikidata_search.find_props_by_name(search_text)
+    serialized_payload = serializer(search_payload)
+    return jsonify(serialized_payload)

sand/extension_interface/search.py

@@ -1,23 +1,29 @@
 from abc import ABC, abstractmethod
 from typing import Dict
+from sand.models.search import SearchPayload
 
 
-class ISearch(ABC):
-    """ Search Interface to support searches from multiple
+class IEntitySearch(ABC):
+    """ Entity Search Interface to support searches from multiple
         KG datastores.
     """
-
     @abstractmethod
-    def find_class_by_name(self, search_text: str) -> Dict:
-        """Search Class using name"""
+    def find_entity_by_name(self, search_text: str) -> SearchPayload:
+        """Search Entity using name"""
         pass
 
+
+class IOntologySearch(ABC):
+    """ Class and Property Ontology Search Interface to support searches from multiple
+        KG datastores.
+    """
+
     @abstractmethod
-    def find_entity_by_name(self, search_text: str) -> Dict:
-        """Search Entity using name"""
+    def find_class_by_name(self, search_text: str) -> SearchPayload:
+        """Search Class using name"""
         pass
 
     @abstractmethod
-    def find_props_by_name(self, search_text: str) -> Dict:
+    def find_props_by_name(self, search_text: str) -> SearchPayload:
         """Search properties using name"""
-        pass
+        pass

sand/extensions/search/wikidata_search.py

@@ -1,17 +1,16 @@
 import requests
-from flask import request, jsonify
 from typing import Dict
 import nh3
-from sand.controllers.search import ISearch
+from sand.extension_interface.search import IEntitySearch, IOntologySearch
 from sand.models.entity import Entity
-from sand.models.ontology import OntClass, OntProperty
+from sand.models.ontology import OntClass, OntProperty, OntClassAR
+from sand.models.search import SearchItem, SearchPayload
 
 
-class WikidataSearch(ISearch):
+class WikidataSearch(IEntitySearch, IOntologySearch):
 
     def __init__(self):
         self.wikidata_url = "https://www.wikidata.org/w/api.php"
-        self.local_class_idsearch_uri = "http://0.0.0.0:5525/api/classes/"
         self.PARAMS = {
             "action": "query",
             "format": "json",
@@ -22,12 +21,6 @@ def __init__(self):
             "srlimit": 10,
             "srprop": "snippet|titlesnippet"
         }
-        self.search_item_template = {
-            "label": "",
-            "id": "",
-            "description": "",
-            "uri": "",
-        }
 
     def get_class_search_params(self, search_text: str) -> Dict:
         """Updates class search parameters for wikidata API"""
@@ -36,10 +29,9 @@ def get_class_search_params(self, search_text: str) -> Dict:
         class_params['srsearch'] = f"haswbstatement:P279 {search_text}"
         return class_params
 
-    def get_local_class_properties(self, id: str) -> Dict:
+    def get_local_class_properties(self, id: str) -> OntClass:
         """Calls local class search API to fetch all class metadata using class ID"""
-        api_data = requests.get(self.local_class_idsearch_uri + str(id))
-        return api_data.json()
+        return OntClassAR()[id]
 
     def get_entity_search_params(self, search_text: str) -> Dict:
         """Updates entity search parameters for wikidata API"""
@@ -55,51 +47,57 @@ def get_props_search_params(self, search_text: str) -> Dict:
         props_params['srsearch'] = search_text
         return props_params
 
-    def find_class_by_name(self, search_text: str) -> Dict:
+    def find_class_by_name(self, search_text: str) -> SearchPayload:
         """
         Uses Wikidata API to search for classes using their name/text.
         Uses local ID based class search to fetch label and description data.
         """
         request_params = self.get_class_search_params(search_text)
         api_data = requests.get(self.wikidata_url, request_params)
         search_items = api_data.json()['query']['search']
-        payload = {"items": []}
+        payload_items = []
         for search_item in search_items:
-            item = self.search_item_template.copy()
-            item['id'] = search_item['title']
-            local_class_props = self.get_local_class_properties(item['id'])
-            item['label'] = local_class_props['label']
-            item['description'] = local_class_props['description']
-            item['uri'] = OntClass.id2uri(item['id'])
-            payload['items'].append(item)
+            local_class_props = self.get_local_class_properties(search_item['title'])
+            item = SearchItem(
+                label=local_class_props.label,
+                id=search_item['title'],
+                description=local_class_props.description,
+                uri=OntClass.id2uri(search_item['title'])
+            )
+            payload_items.append(item)
+        payload = SearchPayload(payload_items)
         return payload
 
-    def find_entity_by_name(self, search_text: str) -> Dict:
+    def find_entity_by_name(self, search_text: str) -> SearchPayload:
         """Uses Wikidata API to search for entities using their name/text."""
         request_params = self.get_entity_search_params(search_text)
         api_data = requests.get(self.wikidata_url, request_params)
         search_items = api_data.json()['query']['search']
-        payload = {"items": []}
+        payload_items = []
         for search_item in search_items:
-            item = self.search_item_template.copy()
-            item['label'] = nh3.clean(search_item['titlesnippet'], tags=set())
-            item['id'] = search_item['title']
-            item['description'] = nh3.clean(search_item['snippet'], tags=set())
-            item['uri'] = Entity.id2uri(item['id'])
-            payload['items'].append(item)
+            item = SearchItem(
+                label=nh3.clean(search_item['titlesnippet'], tags=set()),
+                id=search_item['title'],
+                description=nh3.clean(search_item['snippet'], tags=set()),
+                uri=Entity.id2uri(search_item['title'])
+            )
+            payload_items.append(item)
+        payload = SearchPayload(payload_items)
         return payload
 
-    def find_props_by_name(self, search_text: str) -> Dict:
+    def find_props_by_name(self, search_text: str) -> SearchPayload:
         """Uses Wikidata API to search for properties using their name/text."""
         request_params = self.get_props_search_params(search_text)
         api_data = requests.get(self.wikidata_url, request_params)
         search_items = api_data.json()['query']['search']
-        payload = {"items": []}
+        payload_items = []
         for search_item in search_items:
-            item = self.search_item_template.copy()
-            item['label'] = nh3.clean(search_item['titlesnippet'], tags=set())
-            item['id'] = search_item['title'].split(":")[1]
-            item['description'] = nh3.clean(search_item['snippet'], tags=set())
-            item['uri'] = OntProperty.id2uri(item['id'])
-            payload['items'].append(item)
+            item = SearchItem(
+                label=nh3.clean(search_item['titlesnippet'], tags=set()),
+                id=search_item['title'].split(":")[1],
+                description=nh3.clean(search_item['snippet'], tags=set()),
+                uri=OntProperty.id2uri(search_item['title'].split(":")[1])
+            )
+            payload_items.append(item)
+        payload = SearchPayload(payload_items)
         return payload

sand/models/ontology.py

@@ -24,7 +24,7 @@ def readable_label(self):
     def uri2id(uri: str) -> str:
         """Convert class URI to entity ID."""
         raise NotImplementedError(
-            "The method is set when its store is initialized. Check the call order to ensure `OntClassAR`is called first"
+            "The method is set when its store is initialized. Check the call order to ensure `OntClassAR` is called first"
         )
 
     @staticmethod

sand/models/search.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+from typing import List
+from gena.serializer import get_dataclass_serializer
+
+
+@dataclass
+class SearchItem:
+    """
+    Search Item dataclass to save the values of each search item in a search
+    """
+    label: str
+    id: str
+    description: str
+    uri: str
+
+
+@dataclass
+class SearchPayload:
+    """
+        Search Payload dataclass to hold several search items
+    """
+    items: List[SearchItem]
+