|
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