mandos.model.apis.querying_pubchem_api

Complexity

Total Complexity

51

Size/Duplication

Total Lines	388
Duplicated Lines	0 %

Importance

Changes

0

25 Methods

Rating	Name	Duplication	Size	Complexity
A	QueryingPubchemApi._fetch_external_table()	0	6	1
A	QueryingPubchemApi._fetch_core_data()	0	9	1
A	QueryingPubchemApi._fetch_hierarchy()	0	9	2
A	QueryingPubchemApi.find_inchikey()	0	4	1
A	QueryingPubchemApi.fetch_data()	0	27	2
A	QueryingPubchemApi._fetch_external_linkset()	0	5	1
A	QueryingPubchemApi._get_parent()	0	15	3
A	QueryingPubchemApi._external_table_url()	0	11	1
A	QueryingPubchemApi._fetch_hierarchies()	0	13	3
A	QueryingPubchemApi.fetch_properties()	0	19	4
A	QueryingPubchemApi._fetch_structure_data()	0	8	2
A	QueryingPubchemApi._get_linked_records()	0	10	1
A	QueryingPubchemApi.find_id()	0	9	2
B	QueryingPubchemApi._scrape_cid()	0	43	5
A	QueryingPubchemApi._get_metadata()	0	6	1
A	QueryingPubchemApi._query_json()	0	9	2
A	QueryingPubchemApi._fetch_external_linksets()	0	7	1
A	QueryingPubchemApi._fetch_data()	0	17	2
A	QueryingPubchemApi.__init__()	0	13	1
A	QueryingPubchemApi._tables_to_use()	0	24	2
A	QueryingPubchemApi._hierarchies_to_use()	0	28	3
A	QueryingPubchemApi._fetch_display_data()	0	5	1
A	QueryingPubchemApi._linksets_to_use()	0	6	1
B	QueryingPubchemApi._strip_by_key_in_place()	0	10	7
A	QueryingPubchemApi._fetch_external_tables()	0	7	1

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

import io
import time
from datetime import datetime, timezone
from typing import Any, List, Mapping, Optional, Sequence, Tuple, Union
from urllib.error import HTTPError

import orjson

Unable to import 'orjson'

import pandas as pd

Unable to import 'pandas'

import regex

Unable to import 'regex'

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from pocketutils.core.exceptions import (

Unable to import 'pocketutils.core.exceptions'

    DataIntegrityError,
    DownloadError,
    LookupFailedError,
)
from pocketutils.core.query_utils import QueryExecutor

Unable to import 'pocketutils.core.query_utils'

from mandos.model.apis.pubchem_api import PubchemApi, PubchemCompoundLookupError
from mandos.model.apis.pubchem_support.pubchem_data import PubchemData
from mandos.model.settings import QUERY_EXECUTORS, SETTINGS
from mandos.model.utils.setup import logger


class QueryingPubchemApi(PubchemApi):

Missing class docstring

    def __init__(

Too many arguments (6/5)

        self,

Wrong hanging indentation before block (add 4 spaces).

        chem_data: bool = True,

Wrong hanging indentation before block (add 4 spaces).

        extra_tables: bool = False,

Wrong hanging indentation before block (add 4 spaces).

        classifiers: bool = False,

Wrong hanging indentation before block (add 4 spaces).

        extra_classifiers: bool = False,

Wrong hanging indentation before block (add 4 spaces).

        executor: QueryExecutor = QUERY_EXECUTORS.pubchem,

Wrong hanging indentation before block (add 4 spaces).

    ):
        self._use_chem_data = chem_data
        self._use_extra_tables = extra_tables
        self._use_classifiers = classifiers
        self._use_extra_classifiers = extra_classifiers
        self._executor = executor

    _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 find_inchikey(self, cid: int) -> str:

Missing function or method docstring

        # return self.fetch_data(cid).names_and_identifiers.inchikey
        props = self.fetch_properties(cid)
        return props["InChIKey"]

    def find_id(self, inchikey: str) -> Optional[int]:

Missing function or method docstring

        # we have to scrape to get the parent anyway,
        # so just download it
        # TODO: there's a faster way

TODO and FIXME comments should generally be avoided.

        try:
            return self.fetch_data(inchikey).cid
        except PubchemCompoundLookupError:
            logger.debug(f"Could not find pubchem ID for {inchikey}", exc_info=True)
            return None

    def fetch_properties(self, cid: int) -> Mapping[str, Any]:

Missing function or method docstring

        url = f"{self._pug}/compound/cid/{cid}/JSON"
        #
        try:
            matches: NestedDotDict = self._query_json(url)
        except HTTPError:
            raise PubchemCompoundLookupError(f"Failed finding pubchem compound {cid}")
        props = matches["PC_Compounds"][0]["props"]
        props = {NestedDotDict(p).get("urn.label"): p.get("value") for p in props}

        def _get_val(v):

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

Either all return statements in a function should return an expression, or none of them should.

            v = NestedDotDict(v)
            for t in ["ival", "fval", "sval"]:

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

                if t in v.keys():
                    return v[t]

        props = {k: _get_val(v) for k, v in props.items() if k is not None and v is not None}
        logger.debug(f"DLed properties for {cid}")
        return props

    def fetch_data(self, inchikey: Union[str, int]) -> [PubchemData]:
        # Dear God this is terrible
        # Here are the steps:
        # 1. Download HTML for the InChI key and scrape the CID
        # 2. Download the "display" JSON data from the CID
        # 3. Look for a Parent-type related compound. If it exists, download its display data
        # 4. Download the structural data and append it
        # 5. Download the external table CSVs and append them
        # 6. Download the link sets and append them
        # 7. Download the classifiers (hierarchies) and append them
        # 8. Attach metadata about how we found this.
        # 9. Return the stupid, stupid result as a massive JSON struct.
        logger.info(f"Downloading PubChem data for {inchikey}")
        if isinstance(inchikey, int):
            cid = inchikey
            # note: this might not be the parent
            # that's ok -- we're about to fix that
            inchikey = self.find_inchikey(cid)
            logger.debug(f"Matched CID {cid} to {inchikey}")
        else:
            cid = self._scrape_cid(inchikey)
            logger.debug(f"Matched inchikey {inchikey} to CID {cid} (scraped)")
        stack = []
        data = self._fetch_data(cid, inchikey, stack)
        logger.debug(f"Downloaded raw data for {cid}/{inchikey}")
        data = self._get_parent(cid, inchikey, data, stack)
        return data

    def _scrape_cid(self, inchikey: str) -> int:
        # This is awful
        # Every attempt to get the actual, correct, unique CID corresponding to the inchikey
        # failed with every proper PubChem API
        # We can't use <pug_view>/data/compound/<inchikey> -- we can only use a CID there
        # I found it with a PUG API
        # https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/CID/GJSURZIOUXUGAL-UHFFFAOYSA-N/record/JSON
        # But that returns multiple results!!
        # There's no apparent way to find out which one is real
        # I tried then querying each found CID, getting the display data, and looking at their parents

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

        # Unfortunately, we end up with multiple contradictory parents
        # Plus, that's insanely slow -- we have to get the full JSON data for each parent
        # Every worse -- the PubChem API docs LIE!!
        # Using ?cids_type=parent DOES NOT GIVE THE PARENT compound
        # 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
        # Ultimately, I found that I can get HTML containing the CID from an inchikey
        # From there, we'll just have to download its "display" data and get the parent, then download that data

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

        url = f"https://pubchem.ncbi.nlm.nih.gov/compound/{inchikey}"
        pat = regex.compile(
            r'<meta property="og:url" content="https://pubchem\.ncbi\.nlm\.nih\.gov/compound/(\d+)">',

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

            flags=regex.V1,
        )
        try:
            for i in range(SETTINGS.pubchem_n_tries):

The variable i seems to be unused.

                try:
                    html = self._executor(url)
                except ConnectionAbortedError:
                    logger.warning(f"Connection aborted for {inchikey} [url: {url}]", exc_info=True)
                    continue
        except HTTPError:
            raise PubchemCompoundLookupError(
                f"Failed finding pubchem compound (HTML) from {inchikey} [url: {url}]"
            )
        match = pat.search(html)

The variable html does not seem to be defined in case the for loop on line 140 is not entered. Are you sure this can never be the case?

        if match is None:
            raise DataIntegrityError(
                f"Something is wrong with the HTML from {url}; og:url not found"
            )
        return int(match.group(1))

    def _get_parent(
        self, cid: int, inchikey: str, data: PubchemData, stack: List[Tuple[int, str]]

Wrong hanging indentation before block (add 4 spaces).

    ) -> PubchemData:
        # guard with is not None: we're not caching, so don't do it twice
        p = data.parent_or_none

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

        if p is None:
            logger.info(f"{cid}/{inchikey} is its own parent")
            return data
        try:
            logger.info(f"{cid}/{inchikey} has parent {p}")
            del data
            return self._fetch_data(p, inchikey, stack)
        except HTTPError:
            raise PubchemCompoundLookupError(
                f"Failed finding pubchem parent compound (JSON)"
                f"for cid {p}, child cid {cid}, inchikey {inchikey}"
            )

    def _fetch_data(self, cid: int, inchikey: str, stack: List[Tuple[int, str]]) -> PubchemData:
        when_started = datetime.now(timezone.utc).astimezone()
        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)

        try:
            data = self._fetch_core_data(cid, stack)
        except HTTPError:
            raise PubchemCompoundLookupError(
                f"Failed finding pubchem compound (JSON) from cid {cid}, inchikey {inchikey}"
            )
        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)

        when_finished = datetime.now(timezone.utc).astimezone()
        logger.trace(f"Downloaded {cid} in {t1-t0} s")
        data["meta"] = self._get_metadata(inchikey, when_started, when_finished, t0, t1)
        self._strip_by_key_in_place(data, "DisplayControls")
        stack.append((cid, inchikey))
        logger.trace(f"Stack: {stack}")
        return PubchemData(NestedDotDict(data))

    def _fetch_core_data(self, cid: int, stack: List[Tuple[int, str]]) -> dict:
        return dict(
            record=self._fetch_display_data(cid),
            linked_records=self._get_linked_records(cid, stack),
            structure=self._fetch_structure_data(cid),
            external_tables=self._fetch_external_tables(cid),
            link_sets=self._fetch_external_linksets(cid),
            classifications=self._fetch_hierarchies(cid),
            properties=NestedDotDict(self.fetch_properties(cid)),
        )

    def _get_metadata(self, inchikey: str, started: datetime, finished: datetime, t0: int, t1: int):

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

Too many arguments (6/5)

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

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

        return dict(
            timestamp_fetch_started=started.isoformat(),
            timestamp_fetch_finished=finished.isoformat(),
            from_lookup=inchikey,
            fetch_nanos_taken=str(t1 - t0),
        )

    def _get_linked_records(self, cid: int, stack: List[Tuple[int, str]]) -> NestedDotDict:
        url = f"{self._pug}/compound/cid/{cid}/cids/JSON?cids_type=same_parent_stereo"
        data = self._query_json(url).sub("IdentifierList")
        logger.debug(f"DLed {len(data.get('CID', []))} linked records for {cid}")
        results = {
            "CID": [*data.get("CID", []), *[s for s, _ in stack]],
            "inchikey": [i for _, i in stack],
        }
        logger.debug(f"Linked records are: {results}")
        return NestedDotDict(results)

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

    def _fetch_structure_data(self, cid: int) -> NestedDotDict:
        if not self._use_chem_data:
            return NestedDotDict({})
        url = f"{self._pug}/compound/cid/{cid}/JSON"
        data = self._query_json(url)["PC_Compounds"][0]
        del data["props"]  # redundant with props section in record
        logger.debug(f"DLed structure for {cid}")
        return data

    def _fetch_external_tables(self, cid: int) -> Mapping[str, str]:
        x = {

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)

            ext_table: self._fetch_external_table(cid, ext_table)
            for ext_table in self._tables_to_use.values()
        }
        logger.debug(f"DLed {len(self._tables_to_use)} external tables for {cid}")
        return x

    def _fetch_external_linksets(self, cid: int) -> Mapping[str, str]:
        x = {

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)

            table: self._fetch_external_linkset(cid, table)
            for table in self._linksets_to_use.values()
        }
        logger.debug(f"DLed {len(self._linksets_to_use)} external linksets for {cid}")
        return x

    def _fetch_hierarchies(self, cid: int) -> NestedDotDict:
        build_up = {}
        for hname, hid in self._hierarchies_to_use.items():
            try:
                build_up[hname] = self._fetch_hierarchy(cid, hname, hid)
            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}")
        # 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")
        logger.debug(f"DLed {len(self._hierarchies_to_use)} hierarchies for {cid}")
        return NestedDotDict(build_up)

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

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)

        logger.debug(f"Downloaded table {table} with {len(df)} rows for {cid}")
        return list(df.to_dict(orient="records"))

    def _fetch_external_linkset(self, cid: int, table: str) -> NestedDotDict:
        url = f"{self._link_db}?format=JSON&type={table}&operation=GetAllLinks&id_1={cid}"
        data = self._executor(url)
        logger.debug(f"Downloaded linkset {table} rows for {cid}")
        return NestedDotDict(orjson.loads(data))

    def _fetch_hierarchy(self, cid: int, hname: str, hid: int) -> Sequence[dict]:
        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 (138/100).

        data: Sequence[dict] = orjson.loads(self._executor(url))["Hierarchies"]
        # underneath Hierarchies is a list of Hierarchy
        logger.debug(f"Found data for classifier {hid}, compound {cid}")
        if len(data) == 0:
            raise LookupFailedError(f"Failed getting hierarchy {hid}")
        logger.debug(f"Downloaded hierarchy {hname} ({hid}) for {cid}")
        return data

    @property
    def _tables_to_use(self) -> Mapping[str, str]:
        dct = {
            "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",
            "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",
        }
        if self._use_extra_tables:
            dct.update(
                {
                    "patent:depositor_provided_patent_identifiers": "patent",
                    "bio:rcsb_pdb:protein_bound_3d_structures": "pdb",
                    "related:pubchem:related_compounds_with_annotation": "compound",
                }
            )
        return dct

    @property
    def _linksets_to_use(self) -> Mapping[str, str]:
        return {
            "lit:pubchem:chemical_cooccurrences_in_literature": "ChemicalNeighbor",
            "lit:pubchem:gene_cooccurrences_in_literature": "ChemicalGeneSymbolNeighbor",
            "lit:pubchem:disease_cooccurrences_in_literature": "ChemicalDiseaseNeighbor",
        }

    @property
    def _hierarchies_to_use(self) -> Mapping[str, int]:
        if not self._use_classifiers:
            return {}
        dct = {
            "MeSH Tree": 1,
            "ChEBI Ontology": 2,
            "WHO ATC Classification System": 79,
            "Guide to PHARMACOLOGY Target Classification": 92,
            "ChEMBL Target Tree": 87,
        }
        if self._use_extra_classifiers:
            dct.update(
                {
                    "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,
                    "EPA CPDat Classification": 99,
                    "FDA Pharm Classes": 78,
                    "ChemIDplus": 84,
                }
            )
        return dct

    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_json(self, url: str) -> NestedDotDict:
        data = self._executor(url)
        data = NestedDotDict(orjson.loads(data))
        if "Fault" in data:
            raise DownloadError(
                f"Request failed ({data.get('Code')}) on {url}: {data.get('Message')}"
            )
        logger.trace(f"Queried {url}")
        return data

    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)


__all__ = ["QueryingPubchemApi"]