0
# Utilities
1
2
Helper functions, decorators, and utilities for FUSE filesystem development. These provide context management, version detection, time handling, and other convenience functionality.
3
4
## Capabilities
5
6
### Context Management
7
8
Functions for accessing FUSE context information and controlling the FUSE mount lifecycle.
9
10
```python { .api }
11
def fuse_get_context() -> tuple[int, int, int]:
12
"""
13
Get the current FUSE request context.
14
15
Returns:
16
Tuple of (uid, gid, pid) for the current request:
17
- uid: User ID of the calling process
18
- gid: Group ID of the calling process
19
- pid: Process ID of the calling process
20
"""
21
22
def fuse_exit() -> None:
23
"""
24
Signal the FUSE main loop to exit.
25
26
This function can be called from within filesystem operations
27
to cleanly unmount and shutdown the FUSE filesystem.
28
"""
29
```
30
31
#### Usage Examples
32
33
```python
34
from mfusepy import Operations, fuse_get_context, fuse_exit
35
import os
36
37
class ContextAwareFS(Operations):
38
def getattr(self, path, fh=None):
39
# Get information about the calling process
40
uid, gid, pid = fuse_get_context()
41
42
# Log access information
43
print(f"Process {pid} (uid={uid}, gid={gid}) accessing {path}")
44
45
# Return different attributes based on caller
46
if uid == 0: # root user
47
return {'st_mode': 0o644, 'st_nlink': 1, 'st_size': 1024}
48
else:
49
return {'st_mode': 0o444, 'st_nlink': 1, 'st_size': 512}
50
51
def write(self, path, data, offset, fh):
52
# Check if current user owns the file
53
uid, gid, pid = fuse_get_context()
54
file_uid = self._get_file_owner(path)
55
56
if uid != file_uid and uid != 0:
57
raise FuseOSError(errno.EACCES)
58
59
return self._write_data(path, data, offset)
60
61
def special_shutdown(self, path, fh):
62
"""Example of programmatic shutdown."""
63
uid, gid, pid = fuse_get_context()
64
65
if path == '/shutdown' and uid == 0:
66
# Only allow root to shutdown
67
fuse_exit()
68
69
return 0
70
```
71
72
### Version Detection
73
74
Functions for detecting FUSE library version and capabilities.
75
76
```python { .api }
77
def get_fuse_version(libfuse) -> tuple[int, int]:
78
"""
79
Get the version of the loaded FUSE library.
80
81
Args:
82
libfuse: FUSE library handle (ctypes CDLL object)
83
84
Returns:
85
Tuple of (major_version, minor_version)
86
"""
87
88
# Global version variables (set at import time)
89
fuse_version_major: int # Major version of loaded FUSE library (2 or 3)
90
fuse_version_minor: int # Minor version of loaded FUSE library
91
```
92
93
#### Usage Examples
94
95
```python
96
from mfusepy import get_fuse_version, fuse_version_major, fuse_version_minor
97
import ctypes
98
99
# Check global version info
100
print(f"FUSE version: {fuse_version_major}.{fuse_version_minor}")
101
102
# Version-specific behavior
103
class VersionAwareFS(Operations):
104
def __init__(self):
105
if fuse_version_major >= 3:
106
# Use FUSE 3.x features
107
self.supports_writeback_cache = True
108
self.supports_readdirplus = True
109
else:
110
# FUSE 2.x compatibility mode
111
self.supports_writeback_cache = False
112
self.supports_readdirplus = False
113
114
def init_with_config(self, conn_info, config_3):
115
"""FUSE 3.x initialization with configuration."""
116
if fuse_version_major >= 3:
117
# Configure FUSE 3.x specific options
118
if config_3:
119
config_3.contents.use_ino = 1
120
config_3.contents.readdir_ino = 1
121
```
122
123
### Time Handling
124
125
Utilities for working with FUSE time structures and conversions.
126
127
```python { .api }
128
def time_of_timespec(ts, use_ns: bool = False) -> float:
129
"""
130
Convert a timespec structure to a float timestamp.
131
132
Args:
133
ts: c_timespec structure
134
use_ns: If True, return nanosecond precision timestamp
135
136
Returns:
137
Timestamp as float (seconds since epoch)
138
"""
139
140
def set_st_attrs(st, attrs: dict[str, Any], use_ns: bool = False) -> None:
141
"""
142
Set stat structure attributes from a dictionary.
143
144
Args:
145
st: c_stat structure to populate
146
attrs: Dictionary of attributes to set
147
use_ns: If True, use nanosecond precision for timestamps
148
"""
149
```
150
151
#### Usage Examples
152
153
```python
154
from mfusepy import Operations, time_of_timespec, set_st_attrs, c_timespec, c_stat
155
import time
156
import ctypes
157
158
class TimestampFS(Operations):
159
def utimens(self, path, times=None):
160
"""Update file timestamps."""
161
if times is None:
162
# Use current time
163
now = time.time()
164
atime = mtime = now
165
else:
166
atime, mtime = times
167
168
# Convert to timespec structures
169
atime_ts = c_timespec()
170
atime_ts.tv_sec = int(atime)
171
atime_ts.tv_nsec = int((atime - int(atime)) * 1e9)
172
173
mtime_ts = c_timespec()
174
mtime_ts.tv_sec = int(mtime)
175
mtime_ts.tv_nsec = int((mtime - int(mtime)) * 1e9)
176
177
# Store timestamps
178
self.files[path]['atime'] = atime_ts
179
self.files[path]['mtime'] = mtime_ts
180
181
return 0
182
183
def getattr(self, path, fh=None):
184
"""Get file attributes with proper timestamp handling."""
185
if path not in self.files:
186
raise FuseOSError(errno.ENOENT)
187
188
file_info = self.files[path]
189
190
# Convert timespec to float timestamps
191
atime = time_of_timespec(file_info['atime'], use_ns=True)
192
mtime = time_of_timespec(file_info['mtime'], use_ns=True)
193
194
return {
195
'st_mode': file_info['mode'],
196
'st_nlink': file_info['nlink'],
197
'st_size': file_info['size'],
198
'st_atime': atime,
199
'st_mtime': mtime,
200
'st_ctime': mtime, # Use mtime for ctime
201
}
202
```
203
204
### Platform-Specific Types
205
206
Platform-specific C type definitions that adapt to the underlying system.
207
208
```python { .api }
209
# Platform-specific types (automatically selected based on system)
210
c_dev_t: type # Device ID type
211
c_fsblkcnt_t: type # File system block count type
212
c_fsfilcnt_t: type # File system file count type
213
c_gid_t: type # Group ID type
214
c_uid_t: type # User ID type
215
c_pid_t: type # Process ID type
216
c_mode_t: type # File mode type
217
c_off_t: type # File offset type
218
c_ino_t: type # Inode number type
219
c_nlink_t: type # Link count type
220
c_blkcnt_t: type # Block count type
221
c_blksize_t: type # Block size type
222
223
# Function type aliases for extended attributes
224
setxattr_t: type # Type for setxattr function pointer
225
getxattr_t: type # Type for getxattr function pointer
226
227
# FUSE 3.x specific types
228
fuse_fill_dir_flags: type # Directory filling flags (FUSE 3.x)
229
fuse_fill_dir_t: type # Directory filling function type (FUSE 3.x)
230
fuse_readdir_flags: type # Directory reading flags (FUSE 3.x)
231
fuse_buf_flags: type # Buffer flags for FUSE operations
232
fuse_pollhandle_p: type # Poll handle pointer type
233
234
# Platform-specific file locking
235
c_flock_t: type # File locking structure (varies by platform)
236
237
# Error constants
238
ENOTSUP: int # "Not supported" error code (platform-specific: 45 on BSD, 95 on Linux, 129/134 on Windows)
239
```
240
241
### Logging
242
243
Pre-configured logger instances for FUSE operations.
244
245
```python { .api }
246
log: logging.Logger # Logger for general FUSE operations ("fuse")
247
callback_logger: logging.Logger # Logger for LoggingMixIn callbacks ("fuse.log-mixin")
248
```
249
250
### Platform Detection
251
252
Global variables for platform and library detection.
253
254
```python { .api }
255
_system: str # Platform system name (Windows, Darwin, Linux, etc.)
256
_machine: str # Architecture/machine type (x86_64, arm64, etc.)
257
_libfuse: ctypes.CDLL # The loaded FUSE library handle
258
_libfuse_path: str # Path to the loaded FUSE library
259
```
260
261
#### Usage Examples
262
263
```python
264
from mfusepy import log, callback_logger, Operations
265
import logging
266
267
class LoggedFS(Operations):
268
def __init__(self):
269
# Configure logging levels
270
log.setLevel(logging.INFO)
271
callback_logger.setLevel(logging.DEBUG)
272
273
# Add custom handler
274
handler = logging.StreamHandler()
275
formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
276
handler.setFormatter(formatter)
277
log.addHandler(handler)
278
279
def getattr(self, path, fh=None):
280
log.info(f"Getting attributes for {path}")
281
282
try:
283
attrs = self._get_file_attrs(path)
284
log.debug(f"Retrieved attrs: {attrs}")
285
return attrs
286
except Exception as e:
287
log.error(f"Error getting attributes for {path}: {e}")
288
raise
289
```
290
291
## Utility Patterns
292
293
### Cross-Platform File Handling
294
295
```python
296
from mfusepy import c_dev_t, c_mode_t, c_off_t, ENOTSUP
297
import os
298
import sys
299
300
class CrossPlatformFS(Operations):
301
def __init__(self):
302
# Adapt behavior based on platform
303
self.is_windows = sys.platform.startswith('win')
304
self.is_macos = sys.platform == 'darwin'
305
self.is_linux = sys.platform.startswith('linux')
306
307
def statfs(self, path):
308
"""Get filesystem statistics with platform adaptations."""
309
try:
310
if self.is_windows:
311
# Windows-specific implementation
312
return self._statfs_windows(path)
313
elif self.is_macos:
314
# macOS-specific implementation
315
return self._statfs_macos(path)
316
else:
317
# Linux/Unix implementation
318
return self._statfs_unix(path)
319
except NotImplementedError:
320
raise FuseOSError(ENOTSUP)
321
```
322
323
### Performance Optimization
324
325
```python
326
from mfusepy import Operations, fuse_version_major
327
import threading
328
import time
329
330
class OptimizedFS(Operations):
331
def __init__(self):
332
self.cache = {}
333
self.cache_lock = threading.RLock()
334
self.cache_timeout = 5 # seconds
335
336
# Use FUSE 3.x optimizations when available
337
self.use_fuse3_features = fuse_version_major >= 3
338
339
def _cached_getattr(self, path):
340
"""Cached file attribute lookup."""
341
with self.cache_lock:
342
cached_entry = self.cache.get(path)
343
if cached_entry:
344
attrs, timestamp = cached_entry
345
if time.time() - timestamp < self.cache_timeout:
346
return attrs
347
348
# Cache miss - fetch fresh attributes
349
attrs = self._fetch_file_attrs(path)
350
self.cache[path] = (attrs, time.time())
351
return attrs
352
353
def getattr(self, path, fh=None):
354
return self._cached_getattr(path)
355
356
def read_buf(self, path, bufpp, size, offset, fh):
357
"""Use FUSE 3.x buffer optimization when available."""
358
if self.use_fuse3_features:
359
# Use zero-copy buffer operations
360
return self._read_buf_optimized(path, bufpp, size, offset, fh)
361
else:
362
# Fall back to regular read
363
data = self.read(path, size, offset, fh)
364
return len(data)
365
```