mandos.model.query_utils

Complexity

Total Complexity

82

Size/Duplication

Total Lines	365
Duplicated Lines	0 %

Importance

Changes

0

42 Methods

Rating	Name	Duplication	Size	Complexity
A	JsonNavigatorListOfLists.__rshift__()	0	2	1
A	JsonNavigatorListOfOptionals.__truediv__()	0	7	2
A	Fns.key_is_in()	0	3	2
B	JsonNavigator._go_inside()	0	15	6
A	JsonNavigatorListOfOptionals._go_inside()	0	6	2
A	Fns.extract_group_1()	0	15	4
A	JsonNavigator._filter()	0	8	3
A	Fns.join_nonnulls()	0	7	2
B	Fns.split_and_flatten_nonnulls()	0	14	6
A	JsonNavigatorListOfOptionals._filter()	0	4	1
A	Fns.not_null()	0	3	2
A	Fns.key_is_not_in()	0	3	2
A	JsonNavigatorListOfLists.__truediv__()	0	10	2
A	JsonNavigatorListOfLists._go_inside()	0	8	1
A	JsonNavigatorListOfLists._filter()	0	2	1
A	JsonNavigatorListOfOptionals.to_list()	0	3	1
A	JsonNavigatorSingleOptional.get()	0	3	1
A	Fns.roman_to_arabic()	0	8	1
A	Fns.request_only()	0	12	3
A	JsonNavigator.__mod__()	0	7	3
A	Fns.n_bar_items()	0	6	1
A	JsonNavigator.create()	0	7	2
A	FilterFn.__call__()	0	2	1
A	Fns.construct()	0	11	1
A	JsonNavigatorListOfLists.__floordiv__()	0	2	1
A	Fns.key_equals()	0	3	2
A	Fns.req_is()	0	9	4
A	JsonNavigatorSingleOptional.__floordiv__()	0	4	1
A	JsonNavigatorListOfOptionals.to_set()	0	3	1
A	Fns.lowercase_unless_acronym()	0	7	2
A	Fns.key_does_not_equal()	0	3	2
A	JsonNavigator.get()	0	3	1
A	Fns.require_only()	0	3	1
A	Fns.identity()	0	3	2
A	JsonNavigator.__truediv__()	0	7	2
A	JsonNavigatorListOfOptionals.__floordiv__()	0	4	1
A	JsonNavigator.__floordiv__()	0	2	1
A	JsonNavigator.__rshift__()	0	2	1
A	Fns.has_key()	0	3	2
A	JsonNavigatorListOfOptionals.__rshift__()	0	3	1
A	Fns.split()	0	8	2
A	Fns.split_bars_to_int()	0	6	1

"""
Support classes to help with querying and processing web data.
"""
from __future__ import annotations

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

import numpy as np

Unable to import 'numpy'

Unused numpy imported as np

from pocketutils.tools.base_tools import BaseTools

Unable to import 'pocketutils.tools.base_tools'

from pocketutils.tools.string_tools import StringTools

Unable to import 'pocketutils.tools.string_tools'

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

logger = logging.getLogger("mandos")

empty_frozenset = frozenset([])


@dataclass(frozen=True)

Missing class docstring

class FilterFn:
    keep_if: Callable[[Any], bool]

    def __call__(self, *args, **kwargs):
        return self.keep_if(*args, **kwargs)


class Fns:

Missing class docstring

    @classmethod
    def has_key(cls, key: str) -> FilterFn:

Missing function or method docstring

        return FilterFn(lambda content: key in content)

    @classmethod
    def key_does_not_equal(cls, key: str, disallowed_value: Any) -> FilterFn:

Missing function or method docstring

        return FilterFn(lambda content: content.get(key) != disallowed_value)

    @classmethod
    def key_equals(cls, key: str, allowed_value: Any) -> FilterFn:

Missing function or method docstring

        return FilterFn(lambda content: content.get(key) == allowed_value)

    @classmethod
    def key_is_not_in(cls, key: str, disallowed_values: Set[Any]) -> FilterFn:

Missing function or method docstring

        return FilterFn(lambda content: content.get(key) not in disallowed_values)

    @classmethod
    def key_is_in(cls, key: str, allowed_values: Set[Any]) -> FilterFn:

Missing function or method docstring

        return FilterFn(lambda content: content.get(key) in allowed_values)

    @classmethod
    def construct(cls, tp: Type[T]) -> Callable[[Iterable[Any]], T]:

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

        """
        Function that constructs an instance whose attributes are from the passed list.
        """

        def construct(things: Iterable[str]) -> Optional[str]:
            x = [s.strip() for s in things]

Variable 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 tp(*x)

        return construct

    @classmethod
    def join_nonnulls(cls) -> Callable[[Iterable[str]], Optional[str]]:

Missing function or method docstring

        def opt_join(things: Iterable[str]) -> Optional[str]:
            x = [s.strip() for s in things]

Variable 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 None if len(x) == 0 else ";".join(x)

        return opt_join

    @classmethod
    def split_and_flatten_nonnulls(

Missing function or method docstring

        cls, sep: str, skip_nulls: bool = False

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[Iterable[Union[str, int, float, None]]], Set[str]]:
        def split_flat(things: Iterable[str]) -> Set[str]:
            results = set()
            for thing in things:
                if thing is not None and thing != float("NaN") or not skip_nulls:
                    # let it fail if skip_nulls is False
                    for bit in str(thing).split(sep):
                        results.add(bit.strip())
            return results

        return split_flat

    @classmethod
    def request_only(cls) -> Callable[[Iterable[str]], Optional[str]]:

Missing function or method docstring

        def only_nonreq(things: Iterable[str]) -> Optional[str]:
            things = [s.strip() for s in things]
            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]

        return only_nonreq

    @classmethod
    def roman_to_arabic(

Missing function or method docstring

        cls, min_val: Optional[int] = None, max_val: Optional[int] = None

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[str], int]:
        def roman_to_arabic(s: str) -> int:

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

            return StringTools.roman_to_arabic(s.strip(), min_val=min_val, max_val=max_val)

        return roman_to_arabic

    @classmethod
    def require_only(cls) -> Callable[[Iterable[str]], Optional[str]]:

Missing function or method docstring

        return BaseTools.only

    @classmethod
    def extract_group_1(

Missing function or method docstring

        cls, pattern: Union[str, re.Pattern]

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[Optional[Any]], Optional[str]]:
        pattern = pattern if isinstance(pattern, re.Pattern) else re.compile(pattern)

        def _match(thing: Optional[str]) -> Optional[str]:
            if thing is None:
                return None
            match = pattern.fullmatch(str(thing))
            if match is None:
                return None
            return match.group(1)

        return _match

    @classmethod
    def lowercase_unless_acronym(cls) -> Callable[[str], str]:

Missing function or method docstring

        def lowercase_unless_acronym(s: str) -> str:

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

            s = s.strip()
            return s if s.isupper() else s.lower()

        return lowercase_unless_acronym

    @classmethod
    def split_bars_to_int(cls, sep: str = "||") -> Callable[[Optional[str]], FrozenSet[int]]:

Missing function or method docstring

        def split_bars_to_int(value: str) -> FrozenSet[int]:
            return frozenset([int(x) for x in cls.split(sep)(value)])

        return split_bars_to_int

    @classmethod
    def split(cls, sep: str = "||") -> Callable[[Optional[str]], FrozenSet[str]]:

Missing function or method docstring

        def split_bars(value: str) -> FrozenSet[str]:
            if value is None:
                return empty_frozenset
            return frozenset([s.strip() for s in value.split(sep)])

        return split_bars

    @classmethod
    def n_bar_items(cls, sep: str = "||") -> Callable[[Optional[str]], int]:

Missing function or method docstring

        def n_bar_items(value: str) -> int:
            return len(cls.split(sep)(value))

        return n_bar_items

    @classmethod
    def not_null(cls) -> Callable[[Any], bool]:

Missing function or method docstring

        return lambda value: value is not None

    @classmethod
    def identity(cls) -> Callable[[T], T]:

Missing function or method docstring

        return lambda value: value

    @classmethod
    def req_is(cls, type_, nullable: bool = False, then_convert=None) -> Callable[[str], str]:

Missing function or method docstring

        def req_is(value):
            if not isinstance(value, type_):
                raise ValueError(f"{value} is a {type(value)}, not {type_}")
            return value if then_convert is None else then_convert(value)

        req_is.__name__ = f"req_is_{type_}" + ("_or_null" if nullable else "")
        return req_is


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 Fns.identity
    if isinstance(fn, str):

Unnecessary "else" after "return"

        return Fns.request_only()
    else:
        return fn


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)

@dataclass(frozen=True, eq=True)

Empty class docstring

class AbstractJsonNavigator:
    """"""


@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:
        return JsonNavigatorSingleOptional(conversion(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):

Unnecessary "else" after "return"

            return self._filter(key)
        else:
            return self._go_inside(key)

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

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

Unnecessary "else" after "return"

            return self._filter(keys)
        else:
            return self._go_inside(keys)

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

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

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

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


@dataclass(frozen=True, eq=True)

Missing class docstring

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

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

Missing function or method docstring

        return self.contents

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

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

Wrong hanging indentation before block (add 4 spaces).

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

Unnecessary "else" after "return"

            return self._filter(key)
        else:
            return self._go_inside(key)

    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)

    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 __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:
        return JsonNavigatorListOfLists([[z.get(key) for key in keys] for z in self.contents])

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


__all__ = ["JsonNavigator", "JsonNavigatorListOfLists", "Fns", "FilterFn"]