|
2 | 2 |
|
3 | 3 | import json
|
4 | 4 | import os
|
5 |
| -import pathlib |
6 | 5 | import re
|
7 | 6 | import secrets
|
8 | 7 | import shlex
|
|
16 | 15 | from collections.abc import Callable, Iterator
|
17 | 16 | from datetime import date, datetime
|
18 | 17 | from enum import Enum
|
| 18 | +from pathlib import Path |
19 | 19 | from select import EPOLLHUP, EPOLLIN, epoll
|
20 | 20 | from shutil import which
|
21 | 21 | from typing import TYPE_CHECKING, Any, override
|
@@ -73,7 +73,7 @@ def jsonify(obj: Any, safe: bool = True) -> Any:
|
73 | 73 | return obj.isoformat()
|
74 | 74 | if isinstance(obj, list | set | tuple):
|
75 | 75 | return [jsonify(item, safe) for item in obj]
|
76 |
| - if isinstance(obj, pathlib.Path): |
| 76 | + if isinstance(obj, Path): |
77 | 77 | return str(obj)
|
78 | 78 | if hasattr(obj, "__dict__"):
|
79 | 79 | return vars(obj)
|
@@ -116,7 +116,7 @@ def __init__(
|
116 | 116 | cmd = shlex.split(cmd)
|
117 | 117 |
|
118 | 118 | if cmd:
|
119 |
| - if cmd[0][0] != '/' and cmd[0][:2] != './': # pathlib.Path does not work well |
| 119 | + if cmd[0][0] != '/' and cmd[0][:2] != './': # Path() does not work well |
120 | 120 | cmd[0] = locate_binary(cmd[0])
|
121 | 121 |
|
122 | 122 | self.cmd = cmd
|
@@ -245,7 +245,7 @@ def peak(self, output: str | bytes) -> bool:
|
245 | 245 | except UnicodeDecodeError:
|
246 | 246 | return False
|
247 | 247 |
|
248 |
| - peak_logfile = pathlib.Path(f"{storage['LOG_PATH']}/cmd_output.txt") |
| 248 | + peak_logfile = Path(f"{storage['LOG_PATH']}/cmd_output.txt") |
249 | 249 |
|
250 | 250 | change_perm = False
|
251 | 251 | if peak_logfile.exists() is False:
|
@@ -304,7 +304,7 @@ def execute(self) -> bool:
|
304 | 304 |
|
305 | 305 | # https://stackoverflow.com/questions/4022600/python-pty-fork-how-does-it-work
|
306 | 306 | if not self.pid:
|
307 |
| - history_logfile = pathlib.Path(f"{storage['LOG_PATH']}/cmd_history.txt") |
| 307 | + history_logfile = Path(f"{storage['LOG_PATH']}/cmd_history.txt") |
308 | 308 |
|
309 | 309 | change_perm = False
|
310 | 310 | if history_logfile.exists() is False:
|
@@ -496,7 +496,7 @@ def json_stream_to_structure(configuration_identifier: str, stream: str, target:
|
496 | 496 | return False
|
497 | 497 |
|
498 | 498 | # Try using the stream as a filepath that should be read
|
499 |
| - if raw is None and (path := pathlib.Path(stream)).exists(): |
| 499 | + if raw is None and (path := Path(stream)).exists(): |
500 | 500 | try:
|
501 | 501 | raw = path.read_text()
|
502 | 502 | except Exception as err:
|
|
0 commit comments