pocketutils.misc.resources.Resources._warn_expired() - Code Metrics - Inspection of "feat: loguru; notifiers" - dmyersturnbull/pocketutils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 57139e...393c8e )

by Douglas

created 2021-10-28 02:06 UTC

Resources._warn_expired() A

↳ Parent: pocketutils.misc.resources

Complexity

Conditions

Size

Total Lines	31
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	28
nop	6
dl	0
loc	31
rs	9.208
c	0
b	0
f	0

import logging

import typing
from datetime import datetime, timedelta
from pathlib import Path
from typing import MutableMapping, Optional, Tuple, Union

import orjson


from pocketutils.core import PathLike

from pocketutils.core.chars import Chars

from pocketutils.core.dot_dict import NestedDotDict

from pocketutils.core.exceptions import (

    DirDoesNotExistError,
    FileDoesNotExistError,
    MissingResourceError,
    PathExistsError,
)
from pocketutils.tools.common_tools import CommonTools
from pocketutils.tools.filesys_tools import FilesysTools
from pocketutils.tools.unit_tools import UnitTools

_logger = logging.getLogger("pocketutils")


class Resources:

    def __init__(self, path: PathLike, *, logger=_logger):
        self._dir = Path(path)
        self._logger = logger

    def contains(self, *nodes: PathLike, suffix: Optional[str] = None) -> bool:
        """Returns whether a resource file (or dir) exists."""
        return self.path(*nodes, suffix=suffix).exists()

    def path(self, *nodes: PathLike, suffix: Optional[str] = None, exists: bool = False) -> Path:
        """
        Gets a path of a test resource file under ``resources/``.

        Raises:
            MissingResourceError: If the path is not found
        """
        path = Path(self._dir, "resources", *nodes)
        path = path.with_suffix(path.suffix if suffix is None else suffix)
        if exists and not path.exists():
            raise MissingResourceError(f"Resource {path} missing")
        return path

    def file(self, *nodes: PathLike, suffix: Optional[str] = None) -> Path:
        """
        Gets a path of a test resource file under ``resources/``.

        Raises:
            MissingResourceError: If the path is not found
            PathExistsError: If the path is not a file or symlink to a file or is not readable
        """
        path = self.path(*nodes, suffix=suffix)
        info = FilesysTools.get_info(path)
        if not info.is_file:
            raise PathExistsError(f"Resource {path} is not a file!")
        if not info.is_readable_file:
            raise FileDoesNotExistError(f"Resource {path} is not readable")
        return path

    def dir(self, *nodes: PathLike) -> Path:
        """
        Gets a path of a test resource file under ``resources/``.

        Raises:
            MissingResourceError: If the path is not found and ``not missing_ok``
            PathExistsError: If the path is not a dir, symlink to a dir, or mount,
                             or lacks 'R' or 'X' permissions
        """
        path = self.path(*nodes)
        info = FilesysTools.get_info(path)
        if not info.exists:
            raise DirDoesNotExistError(f"Resource {path} does not exist")
        if not info.is_dir:
            raise PathExistsError(f"Resource {path} is not a directory")
        if info.is_readable_dir:
            raise FileDoesNotExistError(f"Resource {path} is not readable")
        return path

    def a_file(self, *nodes: PathLike, suffixes: Optional[typing.Set[str]] = None) -> Path:
        """
        Gets a path of a test resource file under ``resources/``, ignoring suffix.

        Args:
            nodes: Path nodes under ``resources/``
            suffixes: Set of acceptable suffixes; if None, all are accepted
        """
        path = Path(self._dir, "resources", *nodes)
        options = [
            p
            for p in path.parent.glob(path.stem + "*")
            if p.is_file() and (suffixes is None or p.suffix in suffixes)
        ]
        try:
            return CommonTools.only(options)
        except LookupError:
            raise MissingResourceError(f"Resource {path} missing") from None

    def json(self, *nodes: PathLike, suffix: Optional[str] = None) -> NestedDotDict:
        """Reads a JSON file under ``resources/``."""
        path = self.path(*nodes, suffix=suffix)
        data = orjson.loads(Path(path).read_text(encoding="utf8", errors="strict"))
        return NestedDotDict(data)

    def json_dict(self, *nodes: PathLike, suffix: Optional[str] = None) -> MutableMapping:
        """Reads a JSON file under ``resources/``."""
        path = self.path(*nodes, suffix=suffix)
        data = orjson.loads(Path(path).read_text(encoding="utf8", errors="strict"))
        return data

    def check_expired(
        self,

        path: PathLike,

        max_sec: Union[timedelta, float],

        *,

        warn_expired_fmt: str = "{path_rel} is {delta} out of date [{mod_rel}]",

        warn_unknown_fmt: str = "{path_rel} mod date is unknown [created: {create_rel}]",

    ) -> Optional[bool]:
        """
        Warns and returns True if ``path`` mod date is more than ``max_sec`` in the past.
        Returns None if it could not be determined.

        The formatting strings can refer to any of these (will be empty if unknown):
        - path: Full path
        - name: File/dir name
        - path_rel: Path relative to ``self._dir``, or full path if not under
        - now: formatted current datetime
        - [mod/create]_dt: Formatted mod/creation datetime
        - [mod/create]_rel: Mod/create datetime in terms of offset from now
        - [mod/create]_delta: Formatted timedelta from now
        - [mod/create]_delta_sec: Number of seconds from now (negative if now < mod/create dt)

        Args:
            path: A specific path to check
            max_sec: Max seconds, or a timedelta
            warn_expired_fmt: Formatting string in terms of the variables listed above
            warn_unknown_fmt: Formatting string in terms of the variables listed above

        Returns:
            Whether it is expired, or None if it could not be determined
        """
        path = Path(path)
        limit = max_sec if isinstance(max_sec, timedelta) else timedelta(seconds=max_sec)
        now = datetime.now().astimezone()
        info = FilesysTools.get_info(path)
        if not info.mod_dt and now - info.mod_dt > limit:

            self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_expired_fmt)
            return True
        elif not info.mod_dt and (not info.create_dt or (now - info.create_dt) > limit):
            self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_unknown_fmt)
            return None
        return False

    def _warn_expired(

        self,

        now: datetime,

        mod: Optional[datetime],

        created: Optional[datetime],

        path: Path,

        fmt: Optional[str],

    ):
        if isinstance(fmt, str):
            fmt = fmt.format
        if path.is_relative_to(self._dir):
            path_rel = str(path.relative_to(self._dir))
        else:
            path_rel = str(path)
        now_str, mod_str, mod_rel, mod_delta, mod_delta_sec = self._pretty(now, mod)
        _, create_str, create_rel, create_delta, create_delta_sec = self._pretty(now, created)
        msg = fmt(
            path=path,
            path_rel=path_rel,
            name=path.name,
            now=now_str,
            mod_dt=mod_str,
            mod_rel=mod_rel,
            mod_delta=mod_delta,
            mod_sec=mod_delta_sec,
            create_dt=create_str,
            create_rel=create_rel,
            create_delta=create_delta,
            create_sec=create_delta_sec,
        )
        self._logger.warning(msg)

    def _pretty(self, now: datetime, then: Optional[datetime]) -> Tuple[str, str, str, str, str]:
class Foo:
    def some_method(self, x, y):
        return x + y;
        now_str = now.strftime("%Y-%m-%d %H:%M:%S")
        delta = now - then
        if then is None:
            return now_str, "", "", "", ""
        then_str = then.strftime("%Y-%m-%d %H:%M:%S")
        then_rel = UnitTools.approx_time_wrt(now, then)
        delta_str = UnitTools.delta_time_to_str(delta, space=Chars.narrownbsp)
        return now_str, then_str, then_rel, delta_str, str(delta.total_seconds())


__all__ = ["Resources"]


1			import logging
			0 ignored issues – show introduced 2021-10-11 02:12 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			import typing
3			from datetime import datetime, timedelta
4			from pathlib import Path
5			from typing import MutableMapping, Optional, Tuple, Union
6
7			import orjson
			0 ignored issues – show introduced 2021-10-11 02:12 UTC by Report Bug Copy Issue Report Unable to import 'orjson' Loading history...
8
9			from pocketutils.core import PathLike
			0 ignored issues – show Bug introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report The name `core` does not seem to exist in module `pocketutils`. Loading history... introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Cannot import 'pocketutils.core' due to syntax error 'invalid syntax (<unknown>, line 134)' Loading history...
10			from pocketutils.core.chars import Chars
			0 ignored issues – show Bug introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report The name `core` does not seem to exist in module `pocketutils`. Loading history...
11			from pocketutils.core.dot_dict import NestedDotDict
			0 ignored issues – show Bug introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report The name `core` does not seem to exist in module `pocketutils`. Loading history...
12			from pocketutils.core.exceptions import (
			0 ignored issues – show Bug introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report The name `core` does not seem to exist in module `pocketutils`. Loading history...
13			DirDoesNotExistError,
14			FileDoesNotExistError,
15			MissingResourceError,
16			PathExistsError,
17			)
18			from pocketutils.tools.common_tools import CommonTools
19			from pocketutils.tools.filesys_tools import FilesysTools
20			from pocketutils.tools.unit_tools import UnitTools
21
22			_logger = logging.getLogger("pocketutils")
23
24
25			class Resources:
			0 ignored issues – show introduced 2021-10-11 02:12 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
26			def __init__(self, path: PathLike, *, logger=_logger):
27			self._dir = Path(path)
28			self._logger = logger
29
30			def contains(self, *nodes: PathLike, suffix: Optional[str] = None) -> bool:
31			"""Returns whether a resource file (or dir) exists."""
32			return self.path(*nodes, suffix=suffix).exists()
33
34			def path(self, *nodes: PathLike, suffix: Optional[str] = None, exists: bool = False) -> Path:
35			"""
36			Gets a path of a test resource file under ``resources/``.
37
38			Raises:
39			MissingResourceError: If the path is not found
40			"""
41			path = Path(self._dir, "resources", *nodes)
42			path = path.with_suffix(path.suffix if suffix is None else suffix)
43			if exists and not path.exists():
44			raise MissingResourceError(f"Resource {path} missing")
45			return path
46
47			def file(self, *nodes: PathLike, suffix: Optional[str] = None) -> Path:
48			"""
49			Gets a path of a test resource file under ``resources/``.
50
51			Raises:
52			MissingResourceError: If the path is not found
53			PathExistsError: If the path is not a file or symlink to a file or is not readable
54			"""
55			path = self.path(*nodes, suffix=suffix)
56			info = FilesysTools.get_info(path)
57			if not info.is_file:
58			raise PathExistsError(f"Resource {path} is not a file!")
59			if not info.is_readable_file:
60			raise FileDoesNotExistError(f"Resource {path} is not readable")
61			return path
62
63			def dir(self, *nodes: PathLike) -> Path:
64			"""
65			Gets a path of a test resource file under ``resources/``.
66
67			Raises:
68			MissingResourceError: If the path is not found and ``not missing_ok``
69			PathExistsError: If the path is not a dir, symlink to a dir, or mount,
70			or lacks 'R' or 'X' permissions
71			"""
72			path = self.path(*nodes)
73			info = FilesysTools.get_info(path)
74			if not info.exists:
75			raise DirDoesNotExistError(f"Resource {path} does not exist")
76			if not info.is_dir:
77			raise PathExistsError(f"Resource {path} is not a directory")
78			if info.is_readable_dir:
79			raise FileDoesNotExistError(f"Resource {path} is not readable")
80			return path
81
82			def a_file(self, *nodes: PathLike, suffixes: Optional[typing.Set[str]] = None) -> Path:
83			"""
84			Gets a path of a test resource file under ``resources/``, ignoring suffix.
85
86			Args:
87			nodes: Path nodes under ``resources/``
88			suffixes: Set of acceptable suffixes; if None, all are accepted
89			"""
90			path = Path(self._dir, "resources", *nodes)
91			options = [
92			p
93			for p in path.parent.glob(path.stem + "*")
94			if p.is_file() and (suffixes is None or p.suffix in suffixes)
95			]
96			try:
97			return CommonTools.only(options)
98			except LookupError:
99			raise MissingResourceError(f"Resource {path} missing") from None
100
101			def json(self, *nodes: PathLike, suffix: Optional[str] = None) -> NestedDotDict:
102			"""Reads a JSON file under ``resources/``."""
103			path = self.path(*nodes, suffix=suffix)
104			data = orjson.loads(Path(path).read_text(encoding="utf8", errors="strict"))
105			return NestedDotDict(data)
106
107			def json_dict(self, *nodes: PathLike, suffix: Optional[str] = None) -> MutableMapping:
108			"""Reads a JSON file under ``resources/``."""
109			path = self.path(*nodes, suffix=suffix)
110			data = orjson.loads(Path(path).read_text(encoding="utf8", errors="strict"))
111			return data
112
113			def check_expired(
114			self,
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
115			path: PathLike,
			0 ignored issues – show Coding Style introduced 2021-10-11 02:12 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
116			max_sec: Union[timedelta, float],
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
117			*,
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
118			warn_expired_fmt: str = "{path_rel} is {delta} out of date [{mod_rel}]",
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
119			warn_unknown_fmt: str = "{path_rel} mod date is unknown [created: {create_rel}]",
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
120			) -> Optional[bool]:
121			"""
122			Warns and returns True if ``path`` mod date is more than ``max_sec`` in the past.
123			Returns None if it could not be determined.
124
125			The formatting strings can refer to any of these (will be empty if unknown):
126			- path: Full path
127			- name: File/dir name
128			- path_rel: Path relative to ``self._dir``, or full path if not under
129			- now: formatted current datetime
130			- [mod/create]_dt: Formatted mod/creation datetime
131			- [mod/create]_rel: Mod/create datetime in terms of offset from now
132			- [mod/create]_delta: Formatted timedelta from now
133			- [mod/create]_delta_sec: Number of seconds from now (negative if now < mod/create dt)
134
135			Args:
136			path: A specific path to check
137			max_sec: Max seconds, or a timedelta
138			warn_expired_fmt: Formatting string in terms of the variables listed above
139			warn_unknown_fmt: Formatting string in terms of the variables listed above
140
141			Returns:
142			Whether it is expired, or None if it could not be determined
143			"""
144			path = Path(path)
145			limit = max_sec if isinstance(max_sec, timedelta) else timedelta(seconds=max_sec)
146			now = datetime.now().astimezone()
147			info = FilesysTools.get_info(path)
148			if not info.mod_dt and now - info.mod_dt > limit:
			0 ignored issues – show unused-code introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Unnecessary "elif" after "return" Loading history...
149			self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_expired_fmt)
150			return True
151			elif not info.mod_dt and (not info.create_dt or (now - info.create_dt) > limit):
152			self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_unknown_fmt)
153			return None
154			return False
155
156			def _warn_expired(
			0 ignored issues – show best-practice introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history... Comprehensibility introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report This function exceeds the maximum number of variables (18/15). Loading history...
157			self,
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
158			now: datetime,
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
159			mod: Optional[datetime],
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
160			created: Optional[datetime],
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
161			path: Path,
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
162			fmt: Optional[str],
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
163			):
164			if isinstance(fmt, str):
165			fmt = fmt.format
166			if path.is_relative_to(self._dir):
167			path_rel = str(path.relative_to(self._dir))
168			else:
169			path_rel = str(path)
170			now_str, mod_str, mod_rel, mod_delta, mod_delta_sec = self._pretty(now, mod)
171			_, create_str, create_rel, create_delta, create_delta_sec = self._pretty(now, created)
172			msg = fmt(
173			path=path,
174			path_rel=path_rel,
175			name=path.name,
176			now=now_str,
177			mod_dt=mod_str,
178			mod_rel=mod_rel,
179			mod_delta=mod_delta,
180			mod_sec=mod_delta_sec,
181			create_dt=create_str,
182			create_rel=create_rel,
183			create_delta=create_delta,
184			create_sec=create_delta_sec,
185			)
186			self._logger.warning(msg)
187
188			def _pretty(self, now: datetime, then: Optional[datetime]) -> Tuple[str, str, str, str, str]:
			0 ignored issues – show Coding Style introduced 2021-10-28 02:10 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
189			now_str = now.strftime("%Y-%m-%d %H:%M:%S")
190			delta = now - then
191			if then is None:
192			return now_str, "", "", "", ""
193			then_str = then.strftime("%Y-%m-%d %H:%M:%S")
194			then_rel = UnitTools.approx_time_wrt(now, then)
195			delta_str = UnitTools.delta_time_to_str(delta, space=Chars.narrownbsp)
196			return now_str, then_str, then_rel, delta_str, str(delta.total_seconds())
197
198
199			__all__ = ["Resources"]
200

dmyersturnbull / pocketutils

Push — main ( 57139e...393c8e )

Resources._warn_expired() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like