mandos.model.apis.chembl_scrape_api

Complexity

Total Complexity

18

Size/Duplication

Total Lines	150
Duplicated Lines	0 %

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	CachingChemblScrapeApi._fetch_page()	0	10	3
A	_ScraperSingleton.get()	0	5	2
A	QueryingChemblScrapeApi._fetch_page()	0	21	4
A	ChemblScrapeApi._fetch_page()	0	2	1
A	QueryingChemblScrapeApi.__init__()	0	2	1
A	ChemblScrapeApi.fetch_predictions()	0	3	1
A	CachingChemblScrapeApi.__init__()	0	7	1
A	CachingChemblScrapeApi.path()	0	3	1
A	SarPredictionResult.yes_no_mixed()	0	8	1
A	SarPredictionResult.score()	0	8	1

"""
API that web-scrapes ChEMBL.
"""
from __future__ import annotations

import abc
import enum
from pathlib import Path
from typing import Optional, Type

import pandas as pd

Unable to import 'pandas'

from pocketutils.core.query_utils import QueryExecutor

Unable to import 'pocketutils.core.query_utils'

from typeddfs import TypedDfs, TypedDf

Unable to import 'typeddfs'

from mandos.model import Api
from mandos.model.utils import CleverEnum
from mandos.model.settings import QUERY_EXECUTORS, MANDOS_SETTINGS
from mandos.model.utils.scrape import Scraper, By


class SarPredictionResult(CleverEnum):

Missing class docstring

    active = enum.auto()
    inactive = enum.auto()
    empty = enum.auto()
    both = enum.auto()

    @property
    def yes_no_mixed(self) -> str:

Missing function or method docstring

        return {
            SarPredictionResult.active: "yes",
            SarPredictionResult.inactive: "no",
            SarPredictionResult.empty: "mixed",
            SarPredictionResult.both: "mixed",
        }[self]

    @property
    def score(self) -> int:

Missing function or method docstring

        return {
            SarPredictionResult.active: 1,
            SarPredictionResult.inactive: -1,
            SarPredictionResult.empty: 0,
            SarPredictionResult.both: 0,
        }[self]


class ChemblScrapeTable(TypedDf, metaclass=abc.ABCMeta):

Empty class docstring

    """"""


def _parse_conf(df: pd.DataFrame):

Argument 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.copy()
    for t in [70, 80, 90]:

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)

        df[f"confidence_{t}"] = df[f"confidence_{t}"].map(SarPredictionResult.of)


ChemblTargetPredictionTable: TypedDf = (
    TypedDfs.typed("ChemblTargetPredictionTable")
    .subclass(ChemblScrapeTable)
    .require("target_chembl_id", "target_pref_name", "target_organism", dtype=str)
    .require("confidence_70", "confidence_80", "confidence_90", dtype=SarPredictionResult)
    .require("activity_threshold", dtype=float)
    .post(_parse_conf)
).build()


class ChemblScrapePage(CleverEnum):

Missing class docstring

    target_predictions = enum.auto()


class _ScraperSingleton:
    x = None

    @classmethod
    def get(cls, executor: QueryExecutor):

Missing function or method docstring

        if cls.x is None:
            cls.x = Scraper.create(executor)
        return cls.x


class ChemblScrapeApi(Api, metaclass=abc.ABCMeta):

Missing class docstring

    def fetch_predictions(self, cid: str) -> ChemblTargetPredictionTable:

Missing function or method docstring

        return self._fetch_page(
            cid, ChemblScrapePage.target_predictions, ChemblTargetPredictionTable
        )

    def _fetch_page(self, cid: str, page: ChemblScrapePage, table_type: Type[ChemblScrapeTable]):
        raise NotImplementedError()


class QueryingChemblScrapeApi(ChemblScrapeApi):

Missing class docstring

    def __init__(self, executor: QueryExecutor = QUERY_EXECUTORS.chembl):
        self._executor = executor

    def _fetch_page(

Parameters differ from overridden '_fetch_page' method

        self, chembl_id: str, page: ChemblScrapePage, table_type: Type[ChemblScrapeTable]

Wrong hanging indentation before block (add 4 spaces).

    ):
        url = f"https://www.ebi.ac.uk/chembl/embed/#compound_report_card/{chembl_id}/{page}"
        scraper = _ScraperSingleton.get(self._executor)
        scraper.go(url)
        rows = []
        i = 2
        while True:
            table = scraper.find_element("table", By.TAG_NAME)
            for tr in table.find_elements("tr"):

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

                rows += [td.text.strip() for td in tr.find_elements("td")]
            # noinspection PyBroadException
            try:
                scraper.find_elements(str(i), By.LINK_TEXT)
            except Exception:

Catching very general exceptions such as Exception is usually not recommended.

                break
            i += 1
        header = rows[0]
        rows = rows[1:]
        return table_type.of(pd.DataFrame(rows, columns=header))


class CachingChemblScrapeApi(ChemblScrapeApi):

Missing class docstring

    def __init__(
        self,

Wrong hanging indentation before block (add 4 spaces).

        query: Optional[QueryingChemblScrapeApi],

Wrong hanging indentation before block (add 4 spaces).

        cache_dir: Path = MANDOS_SETTINGS.chembl_cache_path,

Wrong hanging indentation before block (add 4 spaces).

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

    def _fetch_page(self, cid: str, page: ChemblScrapePage, table_type: Type[ChemblScrapeTable]):
        path = self.path(cid, page)
        if path.exists():

Unnecessary "elif" after "return"

            return ChemblScrapeTable.read_file(path)
        elif self._query is None:
            return ChemblScrapeTable.new_empty()
        data = self._query._fetch_page(cid, page, table_type)

It seems like _fetch_page was declared protected and should not be accessed from this context.

        path.parent.mkdir(exist_ok=True, parents=True)
        data.write_file(path)
        return data

    def path(self, cid: str, page: ChemblScrapePage):

Missing function or method docstring

        return (self._cache_dir / page.name / cid).with_suffix(
            MANDOS_SETTINGS.archive_filename_suffix
        )


__all__ = [
    "ChemblScrapeApi",
    "ChemblScrapePage",
    "ChemblScrapePage",
    "ChemblTargetPredictionTable",
    "QueryingChemblScrapeApi",
    "CachingChemblScrapeApi",
]