UMLS API client

uts_api_client.py

""" UTS (UMLS Terminology Services) API client """
import json
from pathlib import Path
import requests
from lxml.html import fromstring


class UtsClient:
    """All the UTS REST API requests are handled through this client"""
    def __init__(self, apikey=None):
        if apikey is None:
            file_key = Path(__file__).resolve().parents[1] / 'uts-api-key.txt'
            self.apikey = open(file_key).read().rstrip()
            self.apikey = '####'
        else:
            self.apikey = apikey
        self.service = "http://umlsks.nlm.nih.gov"
        self.headers = {
            "Content-type": "application/x-www-form-urlencoded",
            "Accept": "text/plain",
            "User-Agent": "python"
        }
        self.tgt = None
        self.base_uri = "https://uts-ws.nlm.nih.gov"
        self.version = "current"

    def gettgt(self):
        """Retrieve a ticket granting ticket"""
        auth_uri = "https://utslogin.nlm.nih.gov"
        params = {"apikey": self.apikey}
        auth_endpoint = "/cas/v1/api-key"
        r = requests.post(auth_uri + auth_endpoint, data=params,
                          headers=self.headers)
        response = fromstring(r.text)
        # extract the entire URL needed from the HTML form (action attribute)
        # returned - looks similar to
        # https://utslogin.nlm.nih.gov/cas/v1/tickets/TGT-36471-aYqNLN2rFIJPXKzxwdTNC5ZT7z3B3cTAKfSc5ndHQcUxeaDOLN-cas
        # we make a POST call to this URL in the getst method
        self.tgt = response.xpath('//form/@action')[0]

    def getst(self):
        """Request a single-use service ticket"""
        if self.tgt is None:
            self.gettgt()
        params = {"service": self.service}
        r = requests.post(self.tgt, data=params, headers=self.headers)
        return r.text

    def query_get(self, endpoint, query):
        r = requests.get(self.base_uri+endpoint, params=query)
        if r.status_code == 404:
            return
        return json.loads(r.text)

    def get_concept_mesh_atoms(self, cui, tok):
        endpoint = f"/rest/content/{self.version}/CUI/{cui}/atoms"
        query = {"ticket": self.getst(), "sabs": "MSH"}
        rst = self.query_get(endpoint, query)
        # If MSH term can't be found by CUI, use the defining word to search
        if rst is None:
            endpoint = f"/rest/search/{self.version}"
            query = {"ticket": self.getst(),
                     "string": tok, "searchType": "exact"}
            rst = self.query_get(endpoint, query)
            if rst is not None:
                search_max = 5
                for rec in rst['result']['results']:
                    # search MSH until found
                    endpoint = (f"/rest/content/{self.version}"
                                f"/CUI/{rec['ui']}/atoms")
                    query = {"ticket": self.getst(), "sabs": "MSH"}
                    rst = self.query_get(endpoint, query)
                    if rst is not None:
                        return rst
                    else:
                        search_max -= 1
                    if search_max == 0:
                        return None
        return rst