mandos.model.utils.sources

Complexity

Total Complexity

24

Size/Duplication

Total Lines	99
Duplicated Lines	0 %

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	_Source.__init__()	0	3	1
A	_Source.from_path()	0	9	2
A	JsonSource.get()	0	2	1
A	TextSource.get()	0	2	1
A	_Source._get_txt()	0	7	3
F	_Source.cached_from_url()	0	30	14
A	_Source.from_query()	0	9	1
A	XmlSource.get()	0	2	1

from pathlib import Path

Missing module docstring

from typing import Any, Callable, Mapping, Optional

import decorateme

Unable to import 'decorateme'

import defusedxml.ElementTree as Xml

Unable to import 'defusedxml.ElementTree'

import orjson

Unable to import 'orjson'

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from pocketutils.core.exceptions import LookupFailedError, XValueError

Unable to import 'pocketutils.core.exceptions'

from pocketutils.core.query_utils import QueryExecutor

Unable to import 'pocketutils.core.query_utils'

@decorateme.auto_obj()
class _Source:
    def __init__(self, *getters: Callable[[], Optional[str]], name: str = "unknown"):
        self._getters = getters
        self._name = name

    def _get_txt(self) -> str:
        for getter in self._getters:
            txt = getter()
            if txt is not None:
                return txt
            raise LookupFailedError(
                f"Nothing found for {self._name} {self.__class__.__name__} source"
            )

    @classmethod
    def cached_from_url(

Missing function or method docstring

        cls,

Wrong hanging indentation before block (add 4 spaces).

        path: Optional[Path],

Wrong hanging indentation before block (add 4 spaces).

        url: Optional[str],

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        query: QueryExecutor,

Wrong hanging indentation before block (add 4 spaces).

        query_args: Optional[Mapping[str, Any]] = None,

Wrong hanging indentation before block (add 4 spaces).

    ):
        if path is not None and url is None and query is None:

Unnecessary "elif" after "return"

            return cls.from_path(path)
        elif path is None and url is not None and query is not None:
            return cls.from_query(url, query, query_args)
        elif path is not None and url is not None and query is not None:
            path = Path(path)
            query_args = {} if query_args is None else query_args

            def load():
                return path.read_text(encoding="utf8") if path.exists() else None

            def save():
                data = query(url, encoding="utf8", **query_args)
                if data is not None:
                    if not path.exists():
                        path.write_text(data, encoding="utf8")
                return data

            # noinspection PyArgumentList
            return cls(load, save, name=f"Cache({path}, {url})")
        raise XValueError(f"Cannot create source from path {path}, url {url}, and query {query}")

    @classmethod
    def from_path(cls, path: Path):

Missing function or method docstring

        path = Path(path)

        def one():
            return path.read_text(encoding="utf8") if path.exists() else None

        # noinspection PyArgumentList
        return cls(one, name=f"Cache({path})")

    @classmethod
    def from_query(

Missing function or method docstring

        cls, url: str, query: QueryExecutor, query_args: Optional[Mapping[str, Any]] = None

Wrong hanging indentation before block (add 4 spaces).

    ):
        def two():
            return query(url, encoding="utf8", **query_args)

        # noinspection PyArgumentList
        return cls(two, name=f"Cache({url})")


class TextSource(_Source):

Missing class docstring

    def get(self) -> str:

Missing function or method docstring

        return self._get_txt()


class XmlSource(_Source):

Empty class docstring

    """ """

    def get(self) -> Xml:

Missing function or method docstring

        return Xml.fromstring(self._get_txt())


class JsonSource(_Source):

Empty class docstring

    """ """

    def get(self) -> NestedDotDict:

Missing function or method docstring

        return NestedDotDict(orjson.loads(self._get_txt()))


__all__ = ["TextSource", "XmlSource", "JsonSource"]