mandos.model.pubchem_support._nav_fns

Complexity

Total Complexity

57

Size/Duplication

Total Lines	217
Duplicated Lines	0 %

Importance

Changes

0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	Flatmap.construct()	0	10	1
A	Filter.key_is_in()	0	3	2
A	Filter.has_key()	0	3	2
A	Mapx.extract_group_1()	0	15	4
A	Flatmap.require_only()	0	3	1
A	Mapx.split_to()	0	8	1
A	Mapx.int_date()	0	8	3
A	Filter.key_does_not_equal()	0	3	2
A	Mapx.n_bar_items()	0	6	1
B	Mapx.req_is()	0	11	6
A	Mapx.lowercase_unless_acronym()	0	7	2
B	Mapx.str_to()	0	15	8
A	Mapx.split()	0	10	4
A	Mapx.not_null()	0	3	2
A	Flatmap.join_nonnulls()	0	7	2
A	Flatmap.request_only()	0	13	3
A	Mapx.roman_to_arabic()	0	8	1
A	Filter.key_is_not_in()	0	3	2
A	Filter.key_equals()	0	3	2
B	Mapx.split_and_flatten_nonnulls()	0	14	6
A	Mapx.identity()	0	3	2

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

import logging
import re
from datetime import date, datetime
from typing import (
    Any,
    Callable,
    Set,
    Union,
    Optional,
    Type,
    TypeVar,
    Iterable,
    FrozenSet,
)

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 mandos.model.pubchem_support._nav_model import FilterFn

logger = logging.getLogger("mandos")

T = TypeVar("T")

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

empty_frozenset = frozenset([])


class Filter:

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)


class Flatmap:

Missing class docstring

    @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]:
            return tp(*things)

        return construct

    @classmethod
    def join_nonnulls(cls, sep: str = "; ") -> 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 sep.join(x)

        return opt_join

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

Missing function or method docstring

        def only_nonreq(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]

        return only_nonreq

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

Missing function or method docstring

        return BaseTools.only


class Mapx:

Missing class docstring

    @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 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 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 n_bar_items(cls, sep: str = "||") -> Callable[[Optional[str]], int]:

Missing function or method docstring

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

        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 int_date(cls, nullable: bool = False) -> Callable[[Optional[str]], Optional[date]]:

Missing function or method docstring

        def int_date(s: Optional[str]) -> Optional[date]:

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)

            if s is None and nullable:
                return None
            return datetime.strptime(str(s).strip(), "%Y%m%d").date()

        return int_date

    @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 nullable and value is None:
                pass
            elif not isinstance(value, type_):
                raise TypeError(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

    @classmethod
    def str_to(

Missing function or method docstring

        cls, type_: Callable[[str], T], nullable: bool = False, flex_type: bool = False

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[str], str]:
        def str_to(value: Optional[str]) -> Optional[T]:
            if type_ is not None and not flex_type and not isinstance(value, str):
                raise TypeError(f"{value} is a {type(value)}, not str")
            if not nullable and value is None:

Unnecessary "elif" after "raise"

                raise ValueError(f"Value for type {type_} is None")
            elif value is None:
                return None
            return type_(str(value).strip())

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

    @classmethod
    def split_to(

Missing function or method docstring

        cls, type_, sep: str = "||", nullable: bool = False

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[Optional[str]], FrozenSet[int]]:
        def split_to(value: str) -> FrozenSet[int]:
            return frozenset([type_(x) for x in cls.split(sep, nullable=nullable)(value)])

        return split_to

    @classmethod
    def split(cls, sep: str, nullable: bool = False) -> Callable[[Optional[str]], FrozenSet[str]]:

Missing function or method docstring

        def split(value: str) -> FrozenSet[str]:
            if value is None and not nullable:
                raise ValueError(f"Value is None")

Using an f-string that does not have any interpolated variables

            if value is None:
                return empty_frozenset
            return frozenset([s.strip() for s in value.split(sep)])

        return split


__all__ = ["Filter", "Mapx", "Flatmap"]