mandos.model.apis.pubchem_support._nav._request_only() - Code Metrics - Inspection of "feat: massive changes" - dmyersturnbull/mandos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( ddff4b...7b3fbc )

by Douglas

created 2021-06-30 04:47 UTC

_request_only() A

↳ Parent: mandos.model.apis.pubchem_support._nav

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

from __future__ import annotations

from dataclasses import dataclass
from typing import (
    List,
    Union,
    Sequence,
    Callable,
    Set,
    FrozenSet,
    Any,
    Optional,
    Type,
    TypeVar,
    Iterable,
)

from pocketutils.core.dot_dict import NestedDotDict

from pocketutils.tools.base_tools import BaseTools


from mandos.model.apis.pubchem_support._nav_model import FilterFn


T = TypeVar("T", covariant=True)

V = TypeVar("V", covariant=True)


catchable_errors = (KeyError, ValueError, LookupError, TypeError)


class NavError(Exception):

    pass


class MapError(NavError):

    pass


class FlatmapError(NavError):

    pass


class FilterError(NavError):

    pass


def _identity(x: T) -> T:

    return x


# This happens to be duplicated in nav_utils,
# but I *really* don't want that dependency
def _request_only(things: Iterable[str]) -> Optional[str]:
    # TODO: Did I mean to excludeNone here?

    things = [s.strip() for s in things if s is not None]
    if len(things) > 1:

        raise ValueError(f"{len(things)} items in {things}")
    elif len(things) == 0:
        return None
    else:
        return things[0]


def _get_conversion_fn(fn: Union[None, str, Callable[[Any], Any]]) -> Callable[[Any], Any]:

    if fn is None:
        return _identity
    if isinstance(fn, str):

        return _request_only
    else:
        return fn


@dataclass(frozen=True, eq=True)

class AbstractJsonNavigator:
    """ """


@dataclass(frozen=True, eq=True)

class JsonNavigator(AbstractJsonNavigator):
    contents: List[NestedDotDict]

    @classmethod
    def create(

        cls, dct: Union[dict, NestedDotDict, Sequence[dict], Sequence[NestedDotDict]]

    ) -> JsonNavigator:
        if hasattr(dct, "items"):
            dct = [dct]
        return JsonNavigator([NestedDotDict(dict(**d, _landmark="")) for d in dct])

    @property
    def get(self) -> List[NestedDotDict]:

        return self.contents

    def __truediv__(
        self, key: Union[int, str, FilterFn, Callable[[NestedDotDict], NestedDotDict]]

    ) -> JsonNavigator:
        if isinstance(key, FilterFn):

            try:
                return self._filter(key)
            except catchable_errors as e:

                raise FilterError(f"Failed to go filter navigator with '{key}': {e}")
        else:
            try:
                return self._go_inside(key)
            except catchable_errors as e:

                raise MapError(f"Failed to map navigator with '{key}': {e}")

    def __mod__(self, key: Union[int, str]) -> JsonNavigator:
        new = {}
        for z in self.contents:

            if z[key] in new:
                raise ValueError(f"{key} found twice")
            new[z[key]] = z
        return JsonNavigator([NestedDotDict(new)])

    def __floordiv__(self, keys: Sequence[str]) -> JsonNavigatorListOfLists:
        try:
            return JsonNavigatorListOfLists([[z.get(key) for key in keys] for z in self.contents])
        except catchable_errors as e:

            raise FlatmapError(f"Failed to flatmap from navigator with '{keys}': {e}")

    def __rshift__(self, key: str) -> JsonNavigatorListOfOptionals:
        try:
            return JsonNavigatorListOfOptionals([z.get(key) for z in self.contents])
        except catchable_errors as e:

            raise FlatmapError(f"Failed to 'double-flatmap' from navigator with '{key}': {e}")

    def _filter(self, keep_where: FilterFn) -> JsonNavigator:
        if callable(keep_where):

            return JsonNavigator([z for z in self.contents if keep_where(z)])
        else:
            key, values = keep_where
            if not isinstance(values, (Set, FrozenSet, List)):

                values = {values}
            return JsonNavigator([z for z in self.contents if z.get(key) in values])

    def _go_inside(self, key: Union[int, str]) -> JsonNavigator:
        new = []
        for z in self.contents:

            if key in z:
                # nav = z.get_as("_nav", list, [])
                # nav.append(z[key])
                if isinstance(z.get(key), list):
                    new.extend([NestedDotDict(dict(**m)) for m in z[key]])
                elif isinstance(z.get(key), NestedDotDict):
                    new.append(NestedDotDict(dict(**z[key])))
                elif isinstance(z.get(key), dict):
                    new.append(NestedDotDict(dict(**z[key])))
                else:
                    raise ValueError(f"{key} value is {type(z[key])}: {z[key]}")
        return JsonNavigator(new)


@dataclass(frozen=True, eq=True)

class JsonNavigatorListOfLists(AbstractJsonNavigator):
    contents: List[List[Any]]

    def __truediv__(
        self,

        keys: Union[

            Sequence[Union[None, str, Callable[[Any], Any]]], FilterFn, Callable[[List[T]], Any]
        ],
    ) -> JsonNavigatorListOfLists:
        if isinstance(keys, FilterFn):

            try:
                return self._filter(keys)
            except catchable_errors as e:

                raise FilterError(f"Failed to filter list-of-lists with '{keys}': {e}")
        else:
            try:
                return self._go_inside(keys)
            except catchable_errors as e:

                raise MapError(f"Failed to map list-of-lists with '{keys}': {e}")

    def __rshift__(self, conversion: Callable[[List[List[T]]], Any]) -> JsonNavigatorSingleOptional:
        try:
            return JsonNavigatorSingleOptional(conversion(self.contents))
        except catchable_errors as e:

            raise FlatmapError(
                f"Failed to 'double-flatmap' from list-of-lists with '{conversion}': {e}"
            )

    def __floordiv__(self, conversion: Callable[[List[T]], Any]) -> JsonNavigatorListOfOptionals:
        try:
            return JsonNavigatorListOfOptionals([conversion(z) for z in self.contents])
        except catchable_errors as e:

            raise FlatmapError(f"Failed to flatmap from list-of-lists with '{conversion}': {e}")

    def _filter(self, keep_if: FilterFn) -> JsonNavigatorListOfLists:
        return JsonNavigatorListOfLists([z for z in self.contents if keep_if(z)])

    def _go_inside(
        self, keys: Sequence[Union[None, str, Callable[[Any], Any]]]

    ) -> JsonNavigatorListOfLists:
        fns = [_get_conversion_fn(fn) for fn in keys]
        gotcha = []
        for contents in self.contents:
            got2 = []
            fields_and_funcs = BaseTools.zip_list(contents, fns)
            for i, (value, fn) in enumerate(fields_and_funcs):

                try:
                    got2.append(fn(value))
                except (MapError, ValueError, KeyError, TypeError, AttributeError):
                    raise MapError(
                        f"Failed mapping on field {i}/{len(fields_and_funcs)} (value: {value} ; fn: {fn})"

                        + "\n"
                        + "Full data is: "
                        + str(contents)
                    )
                gotcha.append(got2)
        return JsonNavigatorListOfLists(gotcha)


@dataclass(frozen=True, eq=True)

class JsonNavigatorListOfOptionals(AbstractJsonNavigator):
    contents: List[Optional[T]]

    @property
    def to_list(self) -> List[T]:

        return [k for k in self.contents if k is not None]

    @property
    def to_set(self) -> FrozenSet[T]:

        return frozenset([k for k in self.contents if k is not None])

    def __truediv__(
        self, key: Union[None, str, FilterFn, Callable[[Any], Any]]

    ) -> JsonNavigatorListOfOptionals:
        if isinstance(key, FilterFn):
            try:
                return self._filter(key)
            except catchable_errors as e:

                raise FilterError(f"Failed to filter list-of-optionals with '{key}': {e}")
        else:
            try:
                return self._go_inside(key)
            except (KeyError, ValueError, LookupError) as e:

                raise FlatmapError(f"Failed to map list-of-optionals with '{key}': {e}")

    def __rshift__(self, key: Callable[[List[Optional[Any]]], Any]) -> JsonNavigatorSingleOptional:
        # we can't skip 2
        return self // key

    def __floordiv__(
        self, key: Callable[[List[Optional[Any]]], Any]

    ) -> JsonNavigatorSingleOptional:
        try:
            return JsonNavigatorSingleOptional(key(self.contents))
        except catchable_errors as e:

            raise FlatmapError(f"Failed to flatmap from list-of-optionals with '{key}': {e}")

    def _go_inside(
        self, key: Union[None, str, Callable[[Any], Any]]

    ) -> JsonNavigatorListOfOptionals:
        fn = _get_conversion_fn(key)

        return JsonNavigatorListOfOptionals(
            [None if content is None else fn(content) for content in self.contents]
        )

    def _filter(
        self, keep_if: Union[Callable[[Optional[T]], bool]]

    ) -> JsonNavigatorListOfOptionals:
        return JsonNavigatorListOfOptionals([z for z in self.contents if keep_if(z)])


@dataclass(frozen=True, eq=True)

class JsonNavigatorSingleOptional(AbstractJsonNavigator):
    contents: Optional[T]

    @property
    def get(self) -> T:

        return self.contents

    def __floordiv__(
        self, conversion: Union[Type[T], Callable[[T], V]]

    ) -> JsonNavigatorSingleOptional:
        try:
            return JsonNavigatorSingleOptional(conversion(self.contents))
        except catchable_errors as e:

            raise FlatmapError(f"Failed to map single-optional with '{conversion}': {e}")


__all__ = [
    "JsonNavigator",
    "JsonNavigatorListOfLists",
    "JsonNavigatorListOfOptionals",
    "AbstractJsonNavigator",
    "JsonNavigatorSingleOptional",
    "NavError",
    "MapError",
    "FlatmapError",
    "FilterError",
]


1			from __future__ import annotations
			0 ignored issues – show introduced 2021-02-01 07:03 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			from dataclasses import dataclass
3			from typing import (
4			List,
5			Union,
6			Sequence,
7			Callable,
8			Set,
9			FrozenSet,
10			Any,
11			Optional,
12			Type,
13			TypeVar,
14			Iterable,
15			)
16
17			from pocketutils.core.dot_dict import NestedDotDict
			0 ignored issues – show introduced 2021-06-30 04:51 UTC by Report Bug Copy Issue Report Unable to import 'pocketutils.core.dot_dict' Loading history...
18			from pocketutils.tools.base_tools import BaseTools
			0 ignored issues – show introduced 2021-01-25 23:06 UTC by Report Bug Copy Issue Report Unable to import 'pocketutils.tools.base_tools' Loading history...
19
20			from mandos.model.apis.pubchem_support._nav_model import FilterFn
21
22
23			T = TypeVar("T", covariant=True)
			0 ignored issues – show Coding Style Naming introduced 2021-02-01 07:03 UTC by Report Bug Copy Issue Report Class name "T" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
24			V = TypeVar("V", covariant=True)
			0 ignored issues – show Coding Style Naming introduced 2021-02-01 07:03 UTC by Report Bug Copy Issue Report Class name "V" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
25
26			catchable_errors = (KeyError, ValueError, LookupError, TypeError)
27
28
29			class NavError(Exception):
			0 ignored issues – show introduced 2021-02-01 07:03 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
30			pass
31
32
33			class MapError(NavError):
			0 ignored issues – show introduced 2021-02-01 07:03 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...

dmyersturnbull / mandos

Push — main ( ddff4b...7b3fbc )

_request_only() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like