mandos.model.apis.caching_pubchem_api

Complexity

Total Complexity

17

Size/Duplication

Total Lines	112
Duplicated Lines	0 %

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	CachingPubchemApi._download()	0	26	4
A	CachingPubchemApi.find_similar_compounds()	0	12	2
A	CachingPubchemApi.data_path()	0	2	1
A	CachingPubchemApi.fetch_data()	0	12	3
A	CachingPubchemApi.similarity_path()	0	6	2
A	CachingPubchemApi.__init__()	0	7	1
A	CachingPubchemApi._read_json()	0	4	2
A	CachingPubchemApi._write_siblings()	0	9	2

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

import gzip
import os

The import os seems to be unused.

from pathlib import Path
from typing import FrozenSet, Optional, Union

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.exceptions import XValueError

Unable to import 'pocketutils.core.exceptions'

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


class CachingPubchemApi(PubchemApi):

Missing class docstring

    def __init__(
        self,

Wrong hanging indentation before block (add 4 spaces).

        query: Optional[QueryingPubchemApi],

Wrong hanging indentation before block (add 4 spaces).

        cache_dir: Path = SETTINGS.pubchem_cache_path,

Wrong hanging indentation before block (add 4 spaces).

    ):
        self._cache_dir = cache_dir
        self._query = query

    def fetch_data(self, inchikey_or_cid: Union[str, int]) -> Optional[PubchemData]:

Parameters differ from overridden 'fetch_data' method

        path = self.data_path(inchikey_or_cid)
        if path.exists():
            logger.debug(f"Found cached PubChem data")

Using an f-string that does not have any interpolated variables

            data = self._read_json(path)
            if data is None:
                raise PubchemCompoundLookupError(
                    f"{inchikey_or_cid} previously not found in PubChem"
                )
            self._write_siblings(data)  # TODO: remove

TODO and FIXME comments should generally be avoided.

            return data
        return self._download(inchikey_or_cid)

    def _download(self, inchikey_or_cid: Union[int, str]) -> PubchemData:
        if self._query is None:
            raise PubchemCompoundLookupError(f"{inchikey_or_cid} not cached")
        # logger.debug(f"Downloading PubChem data for {inchikey_or_cid}")
        try:
            data: PubchemData = self._query.fetch_data(inchikey_or_cid)
        except PubchemCompoundLookupError:
            data = PubchemData(NestedDotDict({}))
            path = self.data_path(inchikey_or_cid)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(gzip.compress(data.to_json().encode(encoding="utf8")))
            logger.debug(f"Wrote empty PubChem data to {path}")
            raise
        cid = data.parent_or_self  # if there's ever a parent of a parent, this will NOT work
        path = self.data_path(cid)
        if path.exists():
            logger.debug(f"PubChem data for {inchikey_or_cid} parent CID {cid} exists")
            logger.caution(f"Writing over {path} for {inchikey_or_cid}")
        else:
            logger.debug(f"PubChem data for {inchikey_or_cid} parent CID {cid} does not exist")
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(gzip.compress(data.to_json().encode(encoding="utf8")))
        self._write_siblings(data)
        logger.debug(f"Wrote PubChem data to {path.absolute()}")
        logger.info(f"Got PubChem data for {inchikey_or_cid}")
        return data

    def _write_siblings(self, data: PubchemData):
        cid = data.parent_or_self
        path = self.data_path(cid)
        aliases = {self.data_path(data.inchikey), *data.siblings}
        for sibling in aliases:
            link = self.data_path(sibling)
            link.unlink(missing_ok=True)
            path.link_to(link)
        logger.debug(f"Added aliases {','.join(aliases)} ⇌ {cid} ({path})")

    def data_path(self, inchikey_or_cid: Union[int, str]) -> Path:

Missing function or method docstring

        return self._cache_dir / "data" / f"{inchikey_or_cid}.json.gz"

    def _read_json(self, path: Path) -> Optional[PubchemData]:

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

        deflated = gzip.decompress(path.read_bytes())
        read = orjson.loads(deflated)
        return PubchemData(NestedDotDict(read)) if len(read) > 0 else None

    def similarity_path(self, inchi: str, min_tc: float) -> Path:

Missing function or method docstring

        if not (min_tc * 100).is_integer():
            raise XValueError(f"min_tc {min_tc} is not an increment of 1%")
        percent = int(min_tc * 100)
        path = self._cache_dir / "similarity" / f"{inchi}_{percent}"
        return path.with_suffix(SETTINGS.archive_filename_suffix)

    def find_similar_compounds(self, inchi: str, min_tc: float) -> FrozenSet[int]:
        logger.debug(f"Searching for {inchi} with min TC {min_tc}")
        path = self.similarity_path(inchi, min_tc)
        if path.exists():
            df = pd.read_file(path)

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 frozenset(set(df["cid"].values))
        found = self._query.find_similar_compounds(inchi, min_tc)
        df = pd.DataFrame([pd.Series(dict(cid=cid)) for cid in found])

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)

        path.parent.mkdir(parents=True, exist_ok=True)
        df.write_file(path)
        logger.debug(f"Wrote {len(df)} values for {inchi} with min TC {min_tc}")
        return frozenset(set(df["cid"].values))


__all__ = ["CachingPubchemApi"]