Bulk IDs request

Author: philippesaade-wmde

main.py

@@ -5,6 +5,7 @@
 
 from src.WikidataTextifier import WikidataEntity
 from src.WikidataLabel import WikidataLabel
+from src import utils
 
 # Start Fastapi app
 app = FastAPI(
@@ -48,9 +49,9 @@
         },
     },
 )
-async def get_labels(
+async def get_textified_wd(
     request: Request, background_tasks: BackgroundTasks,
-    id: str = Query(..., examples="Q42"),
+    id: str = Query(..., examples="Q42,Q2"),
     pid: str = Query(None, examples="P31,P279"),
     lang: str = 'en',
     format: str = 'json',
@@ -73,40 +74,56 @@ async def get_labels(
     Returns:
         list: A list of dictionaries containing QIDs and the similarity scores.
     """
-    if not id:
-        response = "ID is missing"
-        return HTTPException(status_code=422, detail=response)
-
     try:
+
+        if not id:
+            response = "ID is missing"
+            return HTTPException(status_code=422, detail=response)
+
         filter_pids = None
         if pid:
             filter_pids = [p.strip() for p in pid.split(',')]
 
-        entity = WikidataEntity.from_id(
-            id,
-            lang=lang,
-            external_ids=external_ids,
-            all_ranks=all_ranks,
-            references=references,
-            filter_pids=filter_pids
-        )
-
-        if not entity:
-            response = "Item not found"
+        qids = [q.strip() for q in id.split(',')]
+        entity_dict = utils.get_wikidata_entities_by_ids(qids)
+
+        if not entity_dict:
+            response = "ID not found"
             return HTTPException(status_code=404, detail=response)
 
-        if format == 'json':
-            results = entity.to_json()
-        elif format == 'triplet':
-            results = entity.to_triplet()
-        elif format == 'text':
-            results = str(entity)
-        else:
-            response = "Invalid format specified"
-            return HTTPException(status_code=422, detail=response)
+        return_data = {}
+        for id in qids:
+            if id in entity_dict:
+                entity = WikidataEntity.from_wd(
+                    entity_dict[id],
+                    id=id,
+                    lang=lang,
+                    external_ids=external_ids,
+                    all_ranks=all_ranks,
+                    references=references,
+                    filter_pids=filter_pids
+                )
+
+                if format == 'text':
+                    results = str(entity)
+                elif format == 'triplet':
+                    results = entity.to_triplet()
+                else:
+                    results = entity.to_json()
+
+                return_data[id] = results
+            else:
+                return_data[id] = None
+
+        if len(qids) == 1:
+            return_data = return_data[qids[0]]
+            if not return_data:
+                response = "Item not found"
+                return HTTPException(status_code=404, detail=response)
 
         background_tasks.add_task(WikidataLabel.delete_old_labels)
-        return results
+        return return_data
+
     except Exception as e:
         traceback.print_exc()
         raise HTTPException(status_code=500, detail="Internal Server Error")

src/WikidataTextifier.py

@@ -467,18 +467,15 @@ class WikidataEntity:
     claims: list[WikidataClaim]
 
     @classmethod
-    def from_id(cls, id: str,
+    def from_wd(cls,
+                entity_dict: dict,
+                id: str,
                 lang: str = 'en',
                 external_ids: bool = True,
                 all_ranks: bool = False,
                 references: bool = False,
                 filter_pids: list[str] | None = None):
 
-        entity_dict = get_wikidata_entities_by_ids(id)
-        if id not in entity_dict:
-            raise ValueError(f"ID not found.")
-
-        entity_dict = entity_dict[id]
         if 'labels' not in entity_dict:
             return None
 
@@ -487,7 +484,10 @@ def from_id(cls, id: str,
 
         aliases = entity_dict['aliases'].get(lang, []) + \
                         entity_dict['aliases'].get('mul', [])
-        aliases = list(set([alias.get('value') for alias in aliases]))
+        aliases = list(set([alias.get('value') \
+                            if isinstance(alias, dict) \
+                                else alias \
+                                    for alias in aliases]))
 
         lazylabel = LazyLabelFactory(lang=lang)
 

src/utils.py

@@ -8,6 +8,41 @@
 import requests
 from .WikidataLabel import WikidataLabel
 
+def get_wikidata_entity_by_id(
+        id,
+        props='labels,descriptions,aliases,statements'
+    ):
+    """
+    Fetches a Wikidata entity by its ID and returns a dictionary of the entity.
+
+    Parameters:
+    - id (str): A Wikidata entity ID (e.g., Q42, P31).
+    - props (str): The properties to retrieve.
+
+    Returns:
+    - dict: A dictionary containing the entity, where keys are entity IDs and values are dictionaries of properties.
+    """
+    entity_type = 'items'
+    if id.startswith('P'):
+        entity_type = 'properties'
+
+    params = {
+        '_fields': props,
+    }
+    headers = {
+        'User-Agent': 'Wikidata Textifier'
+    }
+
+    response = requests.get(
+        f"https://www.wikidata.org/w/rest.php/wikibase/v1/entities/{entity_type}/{id}",
+        params=params,
+        headers=headers
+    )
+    response.raise_for_status()
+    entity_data = response.json()
+    return entity_data
+
+
 def get_wikidata_entities_by_ids(
         ids,
         props='labels|descriptions|aliases|claims'
@@ -56,6 +91,7 @@ def get_wikidata_entities_by_ids(
 
     return entities_data
 
+
 def get_all_missing_labels_ids(data):
     """
     Get the IDs of the entity dictionary where their labels are missing.