Passed
Push — main ( 393c8e...520e83 )
by Douglas
01:36
created

pocketutils.misc.resources.Resources._pretty()   A

Complexity

Conditions 2

Size

Total Lines 9
Code Lines 9

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 2
eloc 9
nop 3
dl 0
loc 9
rs 9.95
c 0
b 0
f 0
1
import logging
0 ignored issues
show
introduced by
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 by
Unable to import 'orjson'
Loading history...
8
9
from pocketutils.core import PathLike
0 ignored issues
show
Bug introduced by
The name core does not seem to exist in module pocketutils.
Loading history...
introduced by
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 by
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 by
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 by
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 by
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 a_toml_file(
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
102
        self, *nodes: PathLike, suffixes: Optional[typing.Set[str]] = None
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
103
    ) -> NestedDotDict:
104
        path = self.a_file(*nodes, suffixes=suffixes)
105
        return NestedDotDict.read_toml(path)
106
107
    def a_json_file(
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
108
        self, *nodes: PathLike, suffixes: Optional[typing.Set[str]] = None
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
109
    ) -> NestedDotDict:
110
        path = self.a_file(*nodes, suffixes=suffixes)
111
        return NestedDotDict.read_json(path)
112
113
    def toml(self, *nodes: PathLike, suffix: Optional[str] = None) -> NestedDotDict:
114
        """Reads a TOML file under ``resources/``."""
115
        path = self.path(*nodes, suffix=suffix)
116
        return NestedDotDict.read_toml(path)
117
118
    def json(self, *nodes: PathLike, suffix: Optional[str] = None) -> NestedDotDict:
119
        """Reads a JSON file under ``resources/``."""
120
        path = self.path(*nodes, suffix=suffix)
121
        return NestedDotDict.read_json(path)
122
123
    def json_dict(self, *nodes: PathLike, suffix: Optional[str] = None) -> MutableMapping:
124
        """Reads a JSON file under ``resources/``."""
125
        path = self.path(*nodes, suffix=suffix)
126
        return orjson.loads(Path(path).read_text(encoding="utf8"))
127
128
    def check_expired(
129
        self,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
130
        path: PathLike,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
131
        max_sec: Union[timedelta, float],
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
132
        *,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
133
        warn_expired_fmt: str = "{path_rel} is {delta} out of date [{mod_rel}]",
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
134
        warn_unknown_fmt: str = "{path_rel} mod date is unknown [created: {create_rel}]",
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
135
    ) -> Optional[bool]:
136
        """
137
        Warns and returns True if ``path`` mod date is more than ``max_sec`` in the past.
138
        Returns None if it could not be determined.
139
140
        The formatting strings can refer to any of these (will be empty if unknown):
141
        - path: Full path
142
        - name: File/dir name
143
        - path_rel: Path relative to ``self._dir``, or full path if not under
144
        - now: formatted current datetime
145
        - [mod/create]_dt: Formatted mod/creation datetime
146
        - [mod/create]_rel: Mod/create datetime in terms of offset from now
147
        - [mod/create]_delta: Formatted timedelta from now
148
        - [mod/create]_delta_sec: Number of seconds from now (negative if now < mod/create dt)
149
150
        Args:
151
            path: A specific path to check
152
            max_sec: Max seconds, or a timedelta
153
            warn_expired_fmt: Formatting string in terms of the variables listed above
154
            warn_unknown_fmt: Formatting string in terms of the variables listed above
155
156
        Returns:
157
            Whether it is expired, or None if it could not be determined
158
        """
159
        path = Path(path)
160
        limit = max_sec if isinstance(max_sec, timedelta) else timedelta(seconds=max_sec)
161
        now = datetime.now().astimezone()
162
        info = FilesysTools.get_info(path)
163
        if info.mod_dt and now - info.mod_dt > limit:
0 ignored issues
show
unused-code introduced by
Unnecessary "elif" after "return"
Loading history...
164
            self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_expired_fmt)
165
            return True
166
        elif not info.mod_dt and (not info.create_dt or (now - info.create_dt) > limit):
167
            self._warn_expired(now, info.mod_dt, info.create_dt, path, warn_unknown_fmt)
168
            return None
169
        return False
170
171
    def _warn_expired(
0 ignored issues
show
Comprehensibility introduced by
This function exceeds the maximum number of variables (18/15).
Loading history...
best-practice introduced by
Too many arguments (6/5)
Loading history...
172
        self,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
173
        now: datetime,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
174
        mod: Optional[datetime],
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
175
        created: Optional[datetime],
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
176
        path: Path,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
177
        fmt: Optional[str],
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
178
    ):
179
        if isinstance(fmt, str):
180
            fmt = fmt.format
181
        if path.is_relative_to(self._dir):
182
            path_rel = str(path.relative_to(self._dir))
183
        else:
184
            path_rel = str(path)
185
        now_str, mod_str, mod_rel, mod_delta, mod_delta_sec = self._pretty(now, mod)
186
        _, create_str, create_rel, create_delta, create_delta_sec = self._pretty(now, created)
187
        msg = fmt(
188
            path=path,
189
            path_rel=path_rel,
190
            name=path.name,
191
            now=now_str,
192
            mod_dt=mod_str,
193
            mod_rel=mod_rel,
194
            mod_delta=mod_delta,
195
            mod_sec=mod_delta_sec,
196
            create_dt=create_str,
197
            create_rel=create_rel,
198
            create_delta=create_delta,
199
            create_sec=create_delta_sec,
200
        )
201
        self._logger.warning(msg)
202
203
    def _pretty(self, now: datetime, then: Optional[datetime]) -> Tuple[str, str, str, str, str]:
0 ignored issues
show
Coding Style introduced by
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...
204
        now_str = now.strftime("%Y-%m-%d %H:%M:%S")
205
        if then is None:
206
            return now_str, "", "", "", ""
207
        delta = now - then
208
        then_str = then.strftime("%Y-%m-%d %H:%M:%S")
209
        then_rel = UnitTools.approx_time_wrt(now, then)
210
        delta_str = UnitTools.delta_time_to_str(delta, space=Chars.narrownbsp)
211
        return now_str, then_str, then_rel, delta_str, str(delta.total_seconds())
212
213
214
__all__ = ["Resources"]
215