mandos.pubchem_api

Complexity

Total Complexity

59

Size/Duplication

Total Lines	336
Duplicated Lines	0 %

Importance

Changes

0

24 Methods

Rating	Name	Duplication	Size	Complexity
A	PubchemApi.fetch_data_from_cid()	0	4	1
A	PubchemApi.find_similar_compounds()	0	2	1
A	QueryingPubchemApi.__init__()	0	2	1
A	PubchemApi.fetch_data()	0	2	1
C	QueryingPubchemApi._fetch_cid()	0	31	9
A	QueryingPubchemApi._fetch_display_data()	0	3	1
A	QueryingPubchemApi._external_table_url()	0	11	1
B	QueryingPubchemApi._fetch_hierarchies()	0	44	5
B	QueryingPubchemApi.fetch_data()	0	52	2
A	QueryingPubchemApi._fetch_compound()	0	7	2
A	CachingPubchemApi.fetch_data()	0	16	3
A	QueryingPubchemApi._query_json()	0	6	2
A	CachingPubchemApi._write_json()	0	5	2
B	QueryingPubchemApi._strip_by_key_in_place()	0	10	7
A	QueryingPubchemApi._query_and_type()	0	9	4
A	QueryingPubchemApi.find_similar_compounds()	0	15	3
A	QueryingPubchemApi._fetch_external_table()	0	5	1
A	QueryingPubchemApi._fetch_external_link_set()	0	4	1
A	QueryingPubchemApi._fetch_misc_data()	0	3	1
A	CachingPubchemApi.similarity_path()	0	3	2
A	CachingPubchemApi.find_similar_compounds()	0	19	4
A	CachingPubchemApi.__init__()	0	6	1
A	CachingPubchemApi._read_json()	0	7	2
A	CachingPubchemApi.data_path()	0	3	2

"""
PubChem querying API.
"""
from __future__ import annotations

import abc
import logging
import time
from urllib.error import HTTPError
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Sequence, Union, FrozenSet

import io
import gzip
import orjson

Unable to import 'orjson'

import pandas as pd

Unable to import 'pandas'

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from pocketutils.core.query_utils import QueryExecutor

Unable to import 'pocketutils.core.query_utils'

from mandos import MandosUtils
from mandos.model.pubchem_data import PubchemData

logger = logging.getLogger("mandos")


class PubchemApi(metaclass=abc.ABCMeta):

Missing class docstring

    def fetch_data_from_cid(self, cid: int) -> Optional[PubchemData]:

Missing function or method docstring

        # separated from fetch_data to make it completely clear what an int value means
        # noinspection PyTypeChecker
        return self.fetch_data(cid)

    def fetch_data(self, inchikey: str) -> Optional[PubchemData]:

Missing function or method docstring

        raise NotImplementedError()

    def find_similar_compounds(self, inchi: Union[int, str], min_tc: float) -> FrozenSet[int]:

Missing function or method docstring

        raise NotImplementedError()


class QueryingPubchemApi(PubchemApi):

Missing class docstring

    def __init__(self):
        self._query = QueryExecutor(0.22, 0.25)

    _pug = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"
    _pug_view = "https://pubchem.ncbi.nlm.nih.gov/rest/pug_view"
    _sdg = "https://pubchem.ncbi.nlm.nih.gov/sdq/sdqagent.cgi"
    _classifications = "https://pubchem.ncbi.nlm.nih.gov/classification/cgi/classifications.fcgi"
    _link_db = "https://pubchem.ncbi.nlm.nih.gov/link_db/link_db_server.cgi"

    def fetch_data(self, inchikey: str) -> Optional[PubchemData]:
        logger.info(f"Downloading PubChem data for {inchikey}")

Use lazy % formatting in logging functions

        data = dict(
            meta=dict(
                timestamp_fetch_started=datetime.now(timezone.utc).astimezone().isoformat(),
                from_lookup=inchikey,
            )
        )
        t0 = time.monotonic_ns()

Variable name "t0" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        cid = self._fetch_compound(inchikey)
        if cid is None:
            return None
        data["record"] = self._fetch_display_data(cid)["Record"]
        external_table_names = {
            "related:pubchem:related_compounds_with_annotation": "compound",
            "drug:clinicaltrials.gov:clinical_trials": "clinicaltrials",
            "pharm:pubchem:reactions": "pathwayreaction",
            "uses:cpdat:uses": "cpdat",
            "tox:chemidplus:acute_effects": "chemidplus",
            "dis:ctd:associated_disorders_and_diseases": "ctd_chemical_disease",
            "lit:pubchem:depositor_provided_pubmed_citations": "pubmed",
            "patent:depositor_provided_patent_identifiers": "patent",
            "bio:rcsb_pdb:protein_bound_3d_structures": "pdb",
            "bio:dgidb:drug_gene_interactions": "dgidb",
            "bio:ctd:chemical_gene_interactions": "ctdchemicalgene",
            "bio:drugbank:drugbank_interactions": "drugbank",
            "bio:drugbank:drug_drug_interactions": "drugbankddi",
            "bio:pubchem:bioassay_results": "bioactivity",
        }
        external_link_set_names = {
            "lit:pubchem:chemical_cooccurrences_in_literature": "ChemicalNeighbor",
            "lit:pubchem:gene_cooccurrences_in_literature": "ChemicalGeneSymbolNeighbor",
            "lit:pubchem:disease_cooccurrences_in_literature": "ChemicalDiseaseNeighbor",
        }
        data["external_tables"] = {
            table: self._fetch_external_table(cid, table) for table in external_table_names.values()
        }
        data["link_sets"] = {
            table: self._fetch_external_link_set(cid, table)
            for table in external_link_set_names.values()
        }
        # get index==0 because we only have 1 compound
        data["structure"] = self._fetch_misc_data(cid)["PC_Compounds"][0]
        del [data["structure"]["props"]]  # redundant with props section in record
        data["classifications"] = self._fetch_hierarchies(cid)["hierarchies"]
        t1 = time.monotonic_ns()

Variable name "t1" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        data["meta"]["timestamp_fetch_finished"] = (
            datetime.now(timezone.utc).astimezone().isoformat()
        )
        data["meta"]["fetch_nanos_taken"] = str(t1 - t0)
        self._strip_by_key_in_place(data, "DisplayControls")
        return PubchemData(NestedDotDict(data))

    def find_similar_compounds(self, inchi: Union[int, str], min_tc: float) -> FrozenSet[int]:
        slash = self._query_and_type(inchi)
        req = self._query(
            f"{self._pug}/compound/similarity/{slash}/{inchi}/JSON?Threshold={min_tc}",
            method="post",
        )
        key = orjson.loads(req)["Waiting"]["ListKey"]
        t0 = time.monotonic()

Variable name "t0" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        while time.monotonic() - t0 < 5:
            # it'll wait as needed here
            resp = self._query(f"{self._pug}/compound/listkey/{key}/cids/JSON")
            resp = NestedDotDict(orjson.loads(resp))
            if resp.get("IdentifierList.CID") is not None:
                return frozenset(resp.req_list_as("IdentifierList.CID", int))
        raise TimeoutError(f"Search for {inchi} using key {key} timed out")

    def _fetch_compound(self, inchikey: Union[int, str]) -> Optional[int]:
        cid = self._fetch_cid(inchikey)
        if cid is None:
            return None
        data = dict(record=self._fetch_display_data(cid)["Record"])
        data = PubchemData(NestedDotDict(data))
        return data.parent_or_self

    def _fetch_cid(self, inchikey: str) -> Optional[int]:
        # The PubChem API docs LIE!!
        # Using ?cids_type=parent DOES NOT give the parent
        # Ex: https://pubchem.ncbi.nlm.nih.gov/compound/656832
        # This is cocaine HCl, which has cocaine (446220) as a parent
        # https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid/656832/JSON
        # gives 656832 back again
        # same thing when querying by inchikey
        slash = self._query_and_type(inchikey)
        url = f"{self._pug}/compound/{slash}/JSON"
        data = self._query_json(url)
        logger.error(url)
        found = []
        for match in data["PC_Compounds"]:
            for c in match["props"]:

Variable name "c" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                if (
                    c["urn"]["label"] == "InChIKey"

Wrong hanging indentation before block (add 4 spaces).

                    and c["urn"]["name"] == "Standard"

Wrong hanging indentation before block (add 4 spaces).

                    and c["value"]["sval"] == inchikey

Wrong hanging indentation before block (add 4 spaces).

                ):
                    if match["id"]["id"] not in found:
                        found.append(match["id"]["id"])
        if len(found) == 0:

Unnecessary "elif" after "return"

            return None
        elif len(found) > 1:
            logger.warning(

Use lazy % formatting in logging functions

                f"Found {len(found)} CIDs for {inchikey}: {found}. Using first ({found[0]})."
            )
        found = found[0]["cid"]
        assert isinstance(found, int), f"Type of {found} is {type(found)}"
        return found

    def _fetch_display_data(self, cid: int) -> Optional[NestedDotDict]:
        url = f"{self._pug_view}/data/compound/{cid}/JSON/?response_type=display"
        return self._query_json(url)

    def _fetch_misc_data(self, cid: int) -> Optional[NestedDotDict]:
        url = f"{self._pug}/compound/cid/{cid}/JSON"
        return self._query_json(url)

    def _query_json(self, url: str) -> NestedDotDict:
        data = self._query(url)
        data = NestedDotDict(orjson.loads(data))
        if "Fault" in data:
            raise ValueError(f"Request failed ({data.get('Code')}) on {url}: {data.get('Message')}")
        return data

    def _fetch_external_link_set(self, cid: int, table: str) -> NestedDotDict:
        url = f"{self._link_db}?format=JSON&type={table}&operation=GetAllLinks&id_1={cid}"
        data = self._query(url)
        return NestedDotDict(orjson.loads(data))

    def _fetch_hierarchies(self, cid: int) -> NestedDotDict:
        hids = {
            "MeSH Tree": 1,
            "ChEBI Ontology": 2,
            "KEGG: Phytochemical Compounds": 5,
            "KEGG: Drug": 14,
            "KEGG: USP": 15,
            "KEGG: Major components of natural products": 69,
            "KEGG: Target-based Classification of Drugs": 22,
            "KEGG: OTC drugs": 25,
            "KEGG: Drug Classes": 96,
            "CAMEO Chemicals": 86,
            "WHO ATC Classification System": 79,
            "Guide to PHARMACOLOGY Target Classification": 92,
            "ChEMBL Target Tree": 87,
            "EPA CPDat Classification": 99,
            "FDA Pharm Classes": 78,
            "ChemIDplus": 84,
        }
        build_up = []
        for hid in hids.values():
            url = f"{self._classifications}?format=json&hid={hid}&search_uid_type=cid&search_uid={cid}&search_type=list&response_type=display"

This line is too long as per the coding-style (142/100).

            try:
                data = orjson.loads(self._query(url))
                logger.debug(f"Found data for classifier {hid}, compound {cid}")

Use lazy % formatting in logging functions

                data = data["Hierarchies"]["Hierarchy"]
                if len(data) > 1:
                    logger.warning(

Use lazy % formatting in logging functions

                        f"Multiple hierarchies for classifier {hid}, compound {cid}; using first"
                    )
                    data = data[0]
                elif len(data) == 1:
                    data = data[0]
                else:
                    raise KeyError("Hierarchy")
            except (HTTPError, KeyError, LookupError) as e:

Variable name "e" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                logger.debug(f"No data for classifier {hid}, compound {cid}: {e}")

Use lazy % formatting in logging functions

                data = {}
            build_up.append(data)
        # These list all of the child nodes for each node
        # Some of them are > 1000 items -- they're HUGE
        # We don't expect to need to navigate to children
        self._strip_by_key_in_place(build_up, "ChildID")
        return NestedDotDict(dict(hierarchies=build_up))

    def _fetch_external_table(self, cid: int, table: str) -> Sequence[dict]:
        url = self._external_table_url(cid, table)
        data = self._query(url)
        df: pd.DataFrame = pd.read_csv(io.StringIO(data))

Variable name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        return list(df.T.to_dict().values())

    def _external_table_url(self, cid: int, collection: str) -> str:
        return (
            self._sdg
            + "?infmt=json"
            + "&outfmt=csv"
            + "&query={ download : * , collection : "
            + collection
            + " , where :{ ands :[{ cid : "
            + str(cid)
            + " }]}}"
        ).replace(" ", "%22")

    def _query_and_type(self, inchi: Union[int, str], req_full: bool = False) -> str:

This method could be written as a function/class method.

        allowed = ["cid", "inchi", "smiles"] if req_full else ["cid", "inchi", "inchikey", "smiles"]
        if isinstance(inchi, int):

Unnecessary "else" after "return"

            return f"cid/{inchi}"
        else:
            query_type = MandosUtils.get_query_type(inchi).name.lower()
            if query_type not in allowed:
                raise ValueError(f"Can't query {inchi} with type {query_type}")
            return f"{query_type}/{inchi}"

    def _strip_by_key_in_place(self, data: Union[dict, list], bad_key: str) -> None:
        if isinstance(data, list):
            for x in data:

Variable name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                self._strip_by_key_in_place(x, bad_key)
        elif isinstance(data, dict):
            for k, v in list(data.items()):

Variable name "v" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                if k == bad_key:
                    del data[k]
                elif isinstance(v, (list, dict)):
                    self._strip_by_key_in_place(v, bad_key)


class CachingPubchemApi(PubchemApi):

Missing class docstring

    def __init__(
        self, cache_dir: Path, querier: Optional[QueryingPubchemApi], compress: bool = True

Wrong hanging indentation before block (add 4 spaces).

    ):
        self._cache_dir = cache_dir
        self._querier = querier
        self._compress = compress

    def fetch_data(self, inchikey: str) -> Optional[PubchemData]:
        path = self.data_path(inchikey)
        if path.exists():
            logger.info(f"Found cached PubChem data at {path.absolute()}")

Use lazy % formatting in logging functions

        elif self._querier is None:
            raise LookupError(f"Key {inchikey} not found in cache")
        else:
            logger.info(f"Downloading PubChem data for {inchikey} ...")

Use lazy % formatting in logging functions

            data = self._querier.fetch_data(inchikey)
            path.parent.mkdir(parents=True, exist_ok=True)
            encoded = data.to_json()
            self._write_json(encoded, path)
            logger.info(f"Wrote PubChem data to {path.absolute()}")

Use lazy % formatting in logging functions

            return data
        read = self._read_json(path)
        return PubchemData(read)

    def _write_json(self, encoded: str, path: Path) -> None:
        if self._compress:
            path.write_bytes(gzip.compress(encoded.encode(encoding="utf8")))
        else:
            path.write_text(encoded, encoding="utf8")

    def _read_json(self, path: Path) -> NestedDotDict:
        if self._compress:
            deflated = gzip.decompress(path.read_bytes())
            read = orjson.loads(deflated)
        else:
            read = orjson.loads(path.read_text(encoding="utf8"))
        return NestedDotDict(read)

    def find_similar_compounds(self, inchi: Union[int, str], min_tc: float) -> FrozenSet[int]:
        path = self.similarity_path(inchi)
        if not path.exists():
            df = None

Variable name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            existing = set()
        else:
            df = pd.read_csv(path, sep="\t")

Variable name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            df = df[df["min_tc"] < min_tc]

Variable name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            existing = set(df["cid"].values)
        if len(existing) == 0:

Unnecessary "else" after "return"

            found = self._querier.find_similar_compounds(inchi, min_tc)
            path.parent.mkdir(parents=True, exist_ok=True)
            new_df = pd.DataFrame([pd.Series(dict(cid=cid, min_tc=min_tc)) for cid in found])
            if df is not None:
                new_df = pd.concat([df, new_df])
            new_df.to_csv(path, sep="\t")
            return frozenset(existing.union(found))
        else:
            return frozenset(existing)

    def data_path(self, inchikey: str):

Missing function or method docstring

        ext = ".json.gz" if self._compress else ".json"
        return self._cache_dir / "data" / f"{inchikey}{ext}"

    def similarity_path(self, inchikey: str):

Missing function or method docstring

        ext = ".tab.gz" if self._compress else ".tab"
        return self._cache_dir / "similarity" / f"{inchikey}{ext}"


__all__ = [
    "PubchemApi",
    "CachingPubchemApi",
    "QueryingPubchemApi",
]