This is my attempt at writing a wrapper around Microsoft's research API (evaluate endoint).

My goals were:

  1. learn how to write classes
  • does it look ok overall?
  • should all methods except for download_publications and save be private (start with underscore)?
  1. make it easier to download entities from the API (use one method to get raw json data & processed tabular data; and another method to easily write either format to a file)

I'm not a software engineer and never use classes for my tasks, but I guess it could be useful in cases like this - when I'm downloading and processing data?

how the code could be improved / do you see a better approach overall?



# -*- coding: utf-8 -*-
import logging

def create_logger(name: str):
    """Create logger with DEBUG level & stream handler."""
    logger = logging.getLogger(name)
    sh = logging.StreamHandler()
    formatter = logging.Formatter("%(asctime)s [%(levelname)s]: %(message)s")
    return logger

logger = create_logger("mag-api-wrapper")


# -*- coding: utf-8 -*-
import json
import requests
import pandas as pd
from time import sleep
from .logger import logger

class MAG:
    """Papers retrieved from Microsoft Academic API."""

    ENDPOINT = "https://api.labs.cognitive.microsoft.com/academic/v1.0/evaluate"
    ENTITIES = {
        "Id": "mag_ID",
        "DN": "original_paper_title",
        "Ti": "normalized_title",
        "W": "normalized_words_in_title",
        "AW": "normalized_words_in_abstract",
        "RA": "restored_abstract",
        "IA": "inverted_abstract",
        "AA": "authors",
        "AuId": "author_id",
        "DAuN": "author_name",
        "Y": "year_published",
        "D": "isodate_published",
        "DOI": "DOI",
        "J": "journals",
        "JN": "journal_name",
        "PB": "publisher",
        "ECC": "estimated_citation_count",
        "F": "fields",
        "DFN": "field_of_study",
        "FN": "normalized_field_of_study",

    def __init__(
        expr: str,
        key: str,
        count: int = 1_000,
        offset: int = 0,
        model: str = "latest",
        attr: str = "DN,Ti,W,AW,IA,AA.AuId,AA.DAuN,Y,D,DOI,J.JN,PB,ECC,F.FN",
        self.expr = expr
        self.key = key
        self.count = count
        self.offset = offset
        self.model = model
        self.attr = attr

        self.json_data = None
        self.table_data = None

    def download_publications(self):
        """Download entities."""
        logger.info(f"Calling Microsoft Academic API with the query: {self.expr}")
        records = list(self.yield_records())
        self.json_data = [item["raw"] for item in records]
        self.table_data = (
            pd.DataFrame([item["processed"] for item in records])
            .drop(["prob", "logprob"], axis=1)
        logger.info(f"Downloaded {self.table_data.shape[0]} entries in total.")

    def save(self, tocsv=None, tojson=None):
        """Write fetched data to files."""
        if tocsv is not None and self.table_data is not None:
            self.table_data.to_csv(tocsv, index=False)
        if tojson is not None and self.json_data is not None:
            with open(tojson, "w", encoding="utf-8") as f:
                json.dump(self.json_data, f, ensure_ascii=False, indent=4)

    def fetch(self, url, params):
        """Make a remote call to Microsoft Academic API."""
        return requests.get(url, params).json()

    def restore_abstract(self, abstract):
        """Restore inverted abstract to its original form."""
        words = abstract["InvertedIndex"]
        total_words = abstract["IndexLength"]

        text = []
        for position in range(0, total_words):
            for word, positions in words.items():
                if position in positions:
        return " ".join(text)

    def process(self, entities):
        """Process entities, including unnesting JSON and restoring
        inverted abstracts to their raw form."""
        for item in entities:
            entity = item.copy()
            if "IA" in entity.keys():
                entity["RA"] = self.restore_abstract(entity["IA"])
                del entity["IA"]
            if "AA" in entity.keys():
                entity["DAuN"] = ";".join(item["DAuN"] for item in entity["AA"])
                entity["AuId"] = ";".join(str(item["AuId"]) for item in entity["AA"])
                del entity["AA"]
            if "F" in entity.keys():
                entity["FN"] = ";".join(item["FN"] for item in entity["F"])
                del entity["F"]
            if "J" in entity.keys():
                if isinstance(entity["J"], dict):
                    entity["JN"] = entity["J"]["JN"]
                elif isinstance(entity["J"], list):
                    entity["JN"] = ";".join(item["JN"] for item in entity["J"])
                    entity["JN"] = entity["J"]
                del entity["J"]
            yield {"raw": item, "processed": entity}

    def yield_records(self):
        """Fetch all entities for a given query expression."""
        params = {
            "expr": self.expr,
            "offset": self.offset,
            "count": self.count,
            "attributes": self.attr,
            "model": self.model,
            "subscription-key": self.key,
        downloaded = 0
        while True:
            data = self.fetch(MAG.ENDPOINT, params)
            if data["entities"] == []:
            yield from self.process(data["entities"])
            params["offset"] += self.count
            downloaded += len(data["entities"])
            logger.info(f"fetched {downloaded} entries.")


from .mag import MAG

__version__ = "0.1.0"


>>> from mag import MAG
>>> pubs = MAG(
        expr="And(And(AW='organized', AW='crime', Y=[2000, 2020]), Composite(F.FN='political science'))",
        key="2q3b955bfa210f9aa1a4eq35fa63378c" #dummy key
>>> pubs.download_publications()
>>> pubs.save(tocsv="data.csv")
>>> pubs.save(tojson="data.json")

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.