mandos.model.pubchem_support._nav

Complexity

Total Complexity

55

Size/Duplication

Total Lines	281
Duplicated Lines	7.12 %

Importance

Changes

0

22 Methods

Rating	Name	Duplication	Size	Complexity
A	JsonNavigator._filter()	0	8	3
A	JsonNavigatorListOfLists.__floordiv__()	0	5	2
A	JsonNavigator.__rshift__()	0	5	2
A	JsonNavigatorListOfOptionals._go_inside()	0	6	2
A	JsonNavigatorListOfOptionals.to_list()	0	3	1
A	JsonNavigatorListOfOptionals._filter()	0	4	1
A	JsonNavigatorListOfOptionals.__rshift__()	0	3	1
A	JsonNavigatorListOfLists._go_inside()	0	8	1
A	JsonNavigatorListOfOptionals.__floordiv__()	0	7	2
A	JsonNavigatorSingleOptional.__floordiv__()	0	7	2
B	JsonNavigator._go_inside()	0	15	6
A	JsonNavigatorListOfLists._filter()	0	2	1
A	JsonNavigatorListOfLists.__truediv__()	10	16	4
A	JsonNavigatorListOfOptionals.__truediv__()	0	13	4
A	JsonNavigator.create()	0	7	2
A	JsonNavigator.__truediv__()	10	13	4
A	JsonNavigatorListOfLists.__rshift__()	0	6	2
A	JsonNavigator.get()	0	3	1
A	JsonNavigator.__mod__()	0	7	3
A	JsonNavigatorSingleOptional.get()	0	3	1
A	JsonNavigatorListOfOptionals.to_set()	0	3	1
A	JsonNavigator.__floordiv__()	0	5	2

from __future__ import annotations

Missing module docstring

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

Unable to import 'pocketutils.core.dot_dict'

from pocketutils.tools.base_tools import BaseTools

Unable to import 'pocketutils.tools.base_tools'

from mandos.model.pubchem_support._nav_model import FilterFn


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

Class name "T" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

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

Class name "V" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

catchable_errors = (KeyError, ValueError, LookupError, TypeError)


class NavError(Exception):

Missing class docstring

    pass


class MapError(NavError):

Missing class docstring

    pass


class FlatmapError(NavError):

Missing class docstring

    pass


class FilterError(NavError):

Missing class docstring

    pass


def _identity(x: T) -> T:

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

    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?

TODO and FIXME comments should generally be avoided.

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

Unnecessary "elif" after "raise"

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

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

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

Unnecessary "else" after "return"

        return _request_only
    else:
        return fn


@dataclass(frozen=True, eq=True)

Empty class docstring

class AbstractJsonNavigator:
    """"""


@dataclass(frozen=True, eq=True)

Missing class docstring

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

    @classmethod
    def create(

Missing function or method docstring

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

Wrong hanging indentation before block (add 4 spaces).

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

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

Missing function or method docstring

        return self.contents

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

Wrong hanging indentation before block (add 4 spaces).

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

This code seems to be duplicated in your project.

            try:
                return self._filter(key)
            except catchable_errors 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)

                raise FilterError(f"Failed to go filter navigator with '{key}': {e}")
        else:
            try:
                return self._go_inside(key)
            except catchable_errors 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)

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

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

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

            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:

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)

            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:

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)

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

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

Unnecessary "else" after "return"

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

Second argument of isinstance is not a type

                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:

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

            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)

Missing class docstring

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

    def __truediv__(
        self,

Wrong hanging indentation before block (add 4 spaces).

        keys: Union[

Wrong hanging indentation before block (add 4 spaces).

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

This code seems to be duplicated in your project.

            try:
                return self._filter(keys)
            except catchable_errors 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)

                raise FilterError(f"Failed to filter list-of-lists with '{keys}': {e}")
        else:
            try:
                return self._go_inside(keys)
            except catchable_errors 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)

                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:

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)

            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:

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)

            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]]]

Wrong hanging indentation before block (add 4 spaces).

    ) -> JsonNavigatorListOfLists:
        fns = [_get_conversion_fn(fn) for fn in keys]
        return JsonNavigatorListOfLists(
            [
                [fn(value) for value, fn in BaseTools.zip_list(contents, fns)]
                for contents in self.contents
            ]
        )


@dataclass(frozen=True, eq=True)

Missing class docstring

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

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

Missing function or method docstring

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

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

Missing function or method docstring

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

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

Wrong hanging indentation before block (add 4 spaces).

    ) -> JsonNavigatorListOfOptionals:
        if isinstance(key, FilterFn):
            try:
                return self._filter(key)
            except catchable_errors 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)

                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:

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)

                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]

Wrong hanging indentation before block (add 4 spaces).

    ) -> JsonNavigatorSingleOptional:
        try:
            return JsonNavigatorSingleOptional(key(self.contents))
        except catchable_errors 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)

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

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

Wrong hanging indentation before block (add 4 spaces).

    ) -> JsonNavigatorListOfOptionals:
        fn = _get_conversion_fn(key)

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

        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]]

Wrong hanging indentation before block (add 4 spaces).

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


@dataclass(frozen=True, eq=True)

Missing class docstring

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

    @property
    def get(self) -> T:

Missing function or method docstring

        return self.contents

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

Wrong hanging indentation before block (add 4 spaces).

    ) -> JsonNavigatorSingleOptional:
        try:
            return JsonNavigatorSingleOptional(conversion(self.contents))
        except catchable_errors 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)

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


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