mandos.model.apis.caching_pubchem_api.CachingPubchemApi.similarity_path() - Code Metrics - Inspection of "fix: various" - dmyersturnbull/mandos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 4074a3...6b8d16 )

by Douglas

created 2021-11-02 03:03 UTC

mandos.model.apis.caching_pubchem_api.CachingPubchemApi.similarity_path() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	3
dl	0
loc	6
rs	10
c	0
b	0
f	0

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

import gzip

import os

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


import orjson

import pandas as pd

from pocketutils.core.dot_dict import NestedDotDict

from pocketutils.core.exceptions import XValueError


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):

    def __init__(
        self,

        query: Optional[QueryingPubchemApi],

        cache_dir: Path = SETTINGS.pubchem_cache_path,

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

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

        path = self.data_path(inchikey_or_cid)
        if path.exists():

            logger.debug(f"Found cached PubChem data for {inchikey_or_cid}")
            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

            return data
        else:
            logger.debug(f"Did NOT find cached PubChem data for {inchikey_or_cid}")
        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:
            path = self.data_path(inchikey_or_cid)
            NestedDotDict({}).write_json(path, mkdirs=True)
            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.warning(f"Writing over {path} for {inchikey_or_cid}")
        else:
            logger.debug(f"PubChem data for {inchikey_or_cid} parent CID {cid} does not exist")
        data._data.write_json(path, mkdirs=True)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
        self._write_siblings(data, inchikey_or_cid)
        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, *others: str):
        cid = data.parent_or_self
        path = self.data_path(cid)
        aliases = {self.data_path(data.inchikey), *data.siblings, *others}
        for sibling in aliases:
            link = self.data_path(sibling)
            if link != path and link.resolve() != path.resolve():
                link.unlink(missing_ok=True)
                path.link_to(link)
        logger.debug(f"Added aliases {','.join([str(s) for s in aliases])} ⇌ {cid} ({path})")

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

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

    def _read_json(self, path: Path) -> Optional[PubchemData]:
class Foo:
    def some_method(self, x, y):
        return x + y;
        dot = NestedDotDict.read_json(path)
        return PubchemData(dot) if len(dot) > 0 else None


__all__ = ["CachingPubchemApi"]


1			"""
2			PubChem caching API.
3			"""
4			from __future__ import annotations
5
6			import gzip
			0 ignored issues – show Unused Code introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report The import `gzip` seems to be unused. Loading history...
7			import os
			0 ignored issues – show Unused Code introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report The import `os` seems to be unused. Loading history...
8			from pathlib import Path
9			from typing import FrozenSet, Optional, Union
			0 ignored issues – show Unused Code introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report Unused FrozenSet imported from typing Loading history...
10
11			import orjson
			0 ignored issues – show introduced 2021-03-22 20:05 UTC by Report Bug Copy Issue Report Unable to import 'orjson' Loading history... Unused Code introduced 2021-11-01 19:34 UTC by Report Bug Copy Issue Report The import `orjson` seems to be unused. Loading history...
12			import pandas as pd
			0 ignored issues – show introduced 2021-01-25 23:06 UTC by Report Bug Copy Issue Report Unable to import 'pandas' Loading history... Unused Code introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report Unused pandas imported as pd Loading history...
13			from pocketutils.core.dot_dict import NestedDotDict
			0 ignored issues – show introduced 2021-01-25 23:06 UTC by Report Bug Copy Issue Report Unable to import 'pocketutils.core.dot_dict' Loading history...
14			from pocketutils.core.exceptions import XValueError
			0 ignored issues – show introduced 2021-09-23 03:01 UTC by Report Bug Copy Issue Report Unable to import 'pocketutils.core.exceptions' Loading history... Unused Code introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report Unused XValueError imported from pocketutils.core.exceptions Loading history...
15
16			from mandos.model.apis.pubchem_api import PubchemApi, PubchemCompoundLookupError
17			from mandos.model.apis.pubchem_support.pubchem_data import PubchemData
18			from mandos.model.apis.querying_pubchem_api import QueryingPubchemApi
19			from mandos.model.settings import SETTINGS
20			from mandos.model.utils.setup import logger
21
22
23			class CachingPubchemApi(PubchemApi):
			0 ignored issues – show introduced 2021-01-25 23:06 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
24			def __init__(
25			self,
			0 ignored issues – show Coding Style introduced 2021-06-30 21:07 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
26			query: Optional[QueryingPubchemApi],
			0 ignored issues – show Coding Style introduced 2021-08-07 00:15 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
27			cache_dir: Path = SETTINGS.pubchem_cache_path,
			0 ignored issues – show Coding Style introduced 2021-08-07 00:15 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
28			):
29			self._cache_dir = cache_dir
30			self._query = query
31
32			def fetch_data(self, inchikey_or_cid: Union[str, int]) -> Optional[PubchemData]:
			0 ignored issues – show Bug introduced 2021-08-16 03:00 UTC by Report Bug Copy Issue Report Parameters differ from overridden 'fetch_data' method Loading history...
33			path = self.data_path(inchikey_or_cid)
34			if path.exists():
			0 ignored issues – show unused-code introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
35			logger.debug(f"Found cached PubChem data for {inchikey_or_cid}")
36			data = self._read_json(path)
37			if data is None:
38			raise PubchemCompoundLookupError(
39			f"{inchikey_or_cid} previously not found in PubChem"
40			)
41			self._write_siblings(data) # TODO: remove
			0 ignored issues – show Coding Style introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
42			return data
43			else:
44			logger.debug(f"Did NOT find cached PubChem data for {inchikey_or_cid}")
45			return self._download(inchikey_or_cid)
46
47			def _download(self, inchikey_or_cid: Union[int, str]) -> PubchemData:
48			if self._query is None:
49			raise PubchemCompoundLookupError(f"{inchikey_or_cid} not cached")
50			# logger.debug(f"Downloading PubChem data for {inchikey_or_cid}")
51			try:
52			data: PubchemData = self._query.fetch_data(inchikey_or_cid)
53			except PubchemCompoundLookupError:
54			path = self.data_path(inchikey_or_cid)
55			NestedDotDict({}).write_json(path, mkdirs=True)
56			logger.debug(f"Wrote empty PubChem data to {path}")
57			raise
58			cid = data.parent_or_self # if there's ever a parent of a parent, this will NOT work
59			path = self.data_path(cid)
60			if path.exists():
61			logger.debug(f"PubChem data for {inchikey_or_cid} parent CID {cid} exists")
62			logger.warning(f"Writing over {path} for {inchikey_or_cid}")
63			else:
64			logger.debug(f"PubChem data for {inchikey_or_cid} parent CID {cid} does not exist")
65			data._data.write_json(path, mkdirs=True)
			0 ignored issues – show Coding Style Best Practice introduced 2021-11-02 03:04 UTC by Report Bug Copy Issue Report It seems like `_data` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
66			self._write_siblings(data, inchikey_or_cid)
67			logger.debug(f"Wrote PubChem data to {path.absolute()}")
68			logger.info(f"Got PubChem data for {inchikey_or_cid}")
69			return data
70
71			def _write_siblings(self, data: PubchemData, *others: str):
72			cid = data.parent_or_self
73			path = self.data_path(cid)
74			aliases = {self.data_path(data.inchikey), data.siblings, others}
75			for sibling in aliases:
76			link = self.data_path(sibling)
77			if link != path and link.resolve() != path.resolve():
78			link.unlink(missing_ok=True)
79			path.link_to(link)
80			logger.debug(f"Added aliases {','.join([str(s) for s in aliases])} ⇌ {cid} ({path})")
81
82			def data_path(self, inchikey_or_cid: Union[int, str]) -> Path:
			0 ignored issues – show introduced 2021-01-25 23:06 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
83			return self._cache_dir / "data" / f"{inchikey_or_cid}.json.gz"
84
85			def _read_json(self, path: Path) -> Optional[PubchemData]:
			0 ignored issues – show Coding Style introduced 2021-08-02 23:39 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
86			dot = NestedDotDict.read_json(path)
87			return PubchemData(dot) if len(dot) > 0 else None
88
89
90			__all__ = ["CachingPubchemApi"]
91

dmyersturnbull / mandos

Push — main ( 4074a3...6b8d16 )

mandos.model.apis.caching_pubchem_api.CachingPubchemApi.similarity_path() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like