1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
import logging
import os
from functools import wraps
from subprocess import run
__all__ = [
"unmount",
"rmdir",
"subdirs",
"subfiles"
]
L = logging.getLogger(__name__)
def log(func):
"""Simple decorator for logging function calls."""
@wraps(func)
def _log(*args, **kwargs):
args_repr = [repr(a) for a in args]
kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
params = ", ".join(args_repr + kwargs_repr)
file_name = os.path.basename(func.__code__.co_filename)
file_name = os.path.splitext(file_name)[0]
func_name = func.__name__ if file_name == "__init__" \
else f"{file_name}.{func.__name__}"
L.debug("%s called with %s", func_name, params)
try:
result = func(*args, **kwargs)
L.debug("%s returned %s", func_name, result)
return result
except Exception as e:
L.error("%s raised %r", func_name, e)
raise e
return _log
@log
def unmount(path):
"""Unmount a FUSE filesystem using fusermount.
See also:
https://manpages.debian.org/bullseye/fuse/fusermount.1.en.html
Args:
path (str): Path to the mounted FUSE filesystem.
Returns:
True if the command was successful, False otherwise.
"""
cmd = ["fusermount", "-u", path]
try:
p = run(cmd, capture_output=True, check=False, text=True)
except Exception as e:
L.error("failed to execute command %s: %r", cmd, e)
return False
return not p.returncode
@log
def rm(path):
"""Remove a file.
Args:
path (str): Path to file.
Returns:
True if the command was successful, False otherwise.
"""
try:
os.remove(path)
except OSError as e:
L.error("failed to remove file %s: %r", path, e)
return False
return True
@log
def rmdir(path):
"""Remove a directory.
Args:
path (str): Path to directory.
Returns:
True if the command was successful, False otherwise.
"""
try:
os.rmdir(path)
except OSError as e:
L.error("failed to remove directory %s: %r", path, e)
return False
return True
def subdirs(path):
"""Yield directory names under given path using os.scandir.
See also:
https://docs.python.org/3/library/os.html#os.scandir
"""
with os.scandir(path) as it:
for entry in it:
if not entry.name.startswith(".") and entry.is_dir():
yield entry.name
def subfiles(path):
"""Yield file names under given path using os.scandir.
See also:
https://docs.python.org/3/library/os.html#os.scandir
"""
with os.scandir(path) as it:
for entry in it:
if not entry.name.startswith(".") and entry.is_file():
yield entry.name
|
