tessl/pypi-mfusepy
Ctypes bindings for the high-level API in libfuse 2 and 3
—
Pending
core-classes.mddocs/
Core Classes
The primary classes that form the foundation of mfusepy's filesystem interface. These provide the essential framework for creating and mounting custom filesystems in Python.
Capabilities
FUSE Class
The main interface class that manages the FUSE mount process, handles callback translation between Python and the native FUSE library, and provides mount configuration options.
class FUSE:
"""Lower level interface for FUSE filesystem implementation."""
OPTIONS = (
('foreground', '-f'),
('debug', '-d'),
('nothreads', '-s'),
)
def __init__(self, operations, mountpoint: str, raw_fi: bool = False,
encoding: str = 'utf-8', **kwargs) -> None:
"""
Mount a FUSE filesystem.
Args:
operations: Instance of Operations subclass implementing filesystem behavior
mountpoint: Directory path where filesystem will be mounted
raw_fi: If True, pass raw fuse_file_info structures to operations
encoding: Character encoding for path/name conversions (default: 'utf-8')
**kwargs: Additional FUSE mount options (foreground, debug, nothreads, etc.)
"""Usage Example
from mfusepy import FUSE, Operations
class MyFS(Operations):
def getattr(self, path, fh=None):
# Implement filesystem operations
pass
# Mount with default options
filesystem = MyFS()
fuse = FUSE(filesystem, '/mnt/myfs')
# Mount with custom options
fuse = FUSE(filesystem, '/mnt/myfs',
foreground=True, debug=True,
allow_other=True, raw_fi=True)Operations Base Class
Abstract base class that defines the interface for filesystem operations. Subclass this to implement custom filesystem behavior by overriding the methods corresponding to the filesystem operations you want to support.
class Operations:
"""Base class to be subclassed for FUSE filesystem operations."""
# File and Directory Operations
def getattr(self, path: str, fh: Optional[int] = None) -> dict[str, Any]:
"""
Get file attributes (like stat()).
Args:
path: File path
fh: File handle (optional)
Returns:
Dictionary with stat-like attributes:
- st_mode: File mode (permissions + type)
- st_nlink: Number of hard links
- st_size: File size in bytes
- st_uid: User ID of owner
- st_gid: Group ID of owner
- st_atime: Access time
- st_mtime: Modification time
- st_ctime: Change time
"""
def readdir(self, path: str, fh: int) -> ReadDirResult:
"""
Read directory contents.
Args:
path: Directory path
fh: Directory file handle
Returns:
Iterable of directory entries. Each entry can be:
- str: Just the filename
- tuple[str, dict[str, int], int]: (name, stat_dict, offset)
- tuple[str, int, int]: (name, ino, offset)
"""
def readlink(self, path: str) -> str:
"""Read the target of a symbolic link."""
def mknod(self, path: str, mode: int, dev: int) -> int:
"""Create a file node (regular file, device special, or named pipe)."""
def mkdir(self, path: str, mode: int) -> int:
"""Create a directory."""
def unlink(self, path: str) -> int:
"""Remove a file."""
def rmdir(self, path: str) -> int:
"""Remove a directory."""
def symlink(self, target: str, source: str) -> int:
"""Create a symbolic link."""
def rename(self, old: str, new: str) -> int:
"""Rename a file or directory."""
def link(self, target: str, source: str) -> int:
"""Create a hard link."""
def chmod(self, path: str, mode: int) -> int:
"""Change file permissions."""
def chown(self, path: str, uid: int, gid: int) -> int:
"""Change file ownership."""
def truncate(self, path: str, length: int, fh: Optional[int] = None) -> int:
"""Truncate a file to a specified length."""
def utimens(self, path: str, times: Optional[tuple[int, int]] = None) -> int:
"""Update file timestamps."""
# File I/O Operations
def open(self, path: str, flags: int) -> int:
"""
Open a file.
Args:
path: File path
flags: Open flags (O_RDONLY, O_WRONLY, O_RDWR, etc.)
Returns:
File handle (integer)
"""
def read(self, path: str, size: int, offset: int, fh: int) -> bytes:
"""
Read data from a file.
Args:
path: File path
size: Number of bytes to read
offset: Offset in file to start reading
fh: File handle
Returns:
Bytes read from file
"""
def write(self, path: str, data, offset: int, fh: int) -> int:
"""
Write data to a file.
Args:
path: File path
data: Data to write (bytes)
offset: Offset in file to start writing
fh: File handle
Returns:
Number of bytes written
"""
def flush(self, path: str, fh: int) -> int:
"""Flush cached data to storage."""
def release(self, path: str, fh: int) -> int:
"""Release/close an open file."""
def fsync(self, path: str, datasync: int, fh: int) -> int:
"""Synchronize file contents."""
def create(self, path: str, mode: int, fi=None) -> int:
"""Create and open a new file atomically."""
# Directory Operations
def opendir(self, path: str) -> int:
"""Open a directory for reading."""
def releasedir(self, path: str, fh: int) -> int:
"""Release/close an open directory."""
def fsyncdir(self, path: str, datasync: int, fh: int) -> int:
"""Synchronize directory contents."""
# Extended Attributes
def setxattr(self, path: str, name: str, value: bytes, options, position=0) -> int:
"""Set an extended attribute."""
def getxattr(self, path: str, name: str, position=0) -> bytes:
"""Get an extended attribute."""
def listxattr(self, path: str) -> Iterable[str]:
"""List extended attributes."""
def removexattr(self, path: str, name: str) -> int:
"""Remove an extended attribute."""
# File System Operations
def statfs(self, path: str) -> dict[str, int]:
"""
Get filesystem statistics.
Returns:
Dictionary with filesystem statistics:
- f_bsize: Block size
- f_frsize: Fragment size
- f_blocks: Total blocks
- f_bfree: Free blocks
- f_bavail: Available blocks
- f_files: Total inodes
- f_ffree: Free inodes
- f_favail: Available inodes
"""
def access(self, path: str, amode: int) -> int:
"""Check file access permissions."""
# Lifecycle Operations
def init(self, path: str) -> None:
"""Initialize filesystem (called after mount)."""
def init_with_config(self, conn_info: Optional['fuse_conn_info'],
config_3: Optional['fuse_config']) -> None:
"""Initialize filesystem with FUSE 3.x configuration."""
def destroy(self, path: str) -> None:
"""Cleanup on unmount."""
# Advanced Operations
def bmap(self, path: str, blocksize: int, idx) -> int:
"""Map file block to device block."""
def ioctl(self, path: str, cmd: int, arg, fh: int, flags: int, data) -> int:
"""Device-specific I/O control."""
def lock(self, path: str, fh: int, cmd: int, lock) -> int:
"""POSIX file locking."""
def poll(self, path: str, fh: int, ph, reventsp) -> int:
"""Poll for I/O events."""
def write_buf(self, path: str, buf, offset: int, fh: int) -> int:
"""Write using buffer (FUSE 3.x optimization)."""
def read_buf(self, path: str, bufpp, size: int, offset: int, fh: int) -> int:
"""Read into buffer (FUSE 3.x optimization)."""
def flock(self, path: str, fh: int, op: int) -> int:
"""BSD file locking."""
def fallocate(self, path: str, mode: int, offset: int, size: int, fh: int) -> int:
"""Allocate file space."""
def copy_file_range(self, path_in: str, fh_in: int, offset_in: int,
path_out: str, fh_out: int, offset_out: int,
length: int, flags: int) -> int:
"""
Copy a range of data from one file to another (FUSE 3.x).
Args:
path_in: Source file path
fh_in: Source file handle
offset_in: Offset in source file
path_out: Destination file path
fh_out: Destination file handle
offset_out: Offset in destination file
length: Number of bytes to copy
flags: Copy operation flags
Returns:
Number of bytes copied
"""
def lseek(self, path: str, offset: int, whence: int, fh: int) -> int:
"""
Seek to a position in a file (FUSE 3.x).
Args:
path: File path
offset: Byte offset to seek to
whence: How to interpret offset (os.SEEK_SET, os.SEEK_CUR, os.SEEK_END)
fh: File handle
Returns:
New file position (absolute offset from beginning)
"""FUSE 3.x Operations Example
class ModernFS(Operations):
def copy_file_range(self, path_in, fh_in, offset_in,
path_out, fh_out, offset_out, length, flags):
"""Efficient file copying using copy_file_range."""
try:
# Read from source file
data = self._read_file_data(path_in, length, offset_in)
# Write to destination file
written = self._write_file_data(path_out, data, offset_out)
return written
except IOError:
raise FuseOSError(errno.EIO)
def lseek(self, path, offset, whence, fh):
"""Seek within a file."""
try:
file_obj = self.open_files[fh]
if whence == os.SEEK_SET:
new_pos = offset
elif whence == os.SEEK_CUR:
new_pos = file_obj.tell() + offset
elif whence == os.SEEK_END:
file_size = self._get_file_size(path)
new_pos = file_size + offset
else:
raise FuseOSError(errno.EINVAL)
return new_pos
except (KeyError, IOError):
raise FuseOSError(errno.EBADF)Implementation Example
import os
import errno
from mfusepy import Operations, FuseOSError
class PassthroughFS(Operations):
"""A passthrough filesystem that mirrors another directory."""
def __init__(self, root):
self.root = root
def _full_path(self, partial):
if partial.startswith("/"):
partial = partial[1:]
path = os.path.join(self.root, partial)
return path
def getattr(self, path, fh=None):
full_path = self._full_path(path)
try:
st = os.lstat(full_path)
return {
'st_mode': st.st_mode,
'st_nlink': st.st_nlink,
'st_size': st.st_size,
'st_uid': st.st_uid,
'st_gid': st.st_gid,
'st_atime': st.st_atime,
'st_mtime': st.st_mtime,
'st_ctime': st.st_ctime,
}
except OSError as e:
raise FuseOSError(e.errno)
def readdir(self, path, fh):
full_path = self._full_path(path)
try:
dirents = ['.', '..']
dirents.extend(os.listdir(full_path))
return dirents
except OSError as e:
raise FuseOSError(e.errno)
def read(self, path, length, offset, fh):
full_path = self._full_path(path)
try:
with open(full_path, 'rb') as f:
f.seek(offset)
return f.read(length)
except OSError as e:
raise FuseOSError(e.errno)C Structure Classes
These classes represent the C structures used by the FUSE library. They are primarily used internally but may be exposed when using raw_fi=True mode.
class c_timespec(ctypes.Structure):
"""Time specification structure."""
class c_utimbuf(ctypes.Structure):
"""Time buffer structure for file timestamps."""
class c_stat(ctypes.Structure):
"""File status structure."""
class c_statvfs(ctypes.Structure):
"""File system statistics structure."""
class fuse_file_info(ctypes.Structure):
"""FUSE file information structure."""
class fuse_context(ctypes.Structure):
"""FUSE context information structure."""
class fuse_conn_info(ctypes.Structure):
"""FUSE connection information structure."""
class fuse_config(ctypes.Structure):
"""FUSE configuration structure."""
class fuse_buf(ctypes.Structure):
"""FUSE buffer structure."""
class fuse_bufvec(ctypes.Structure):
"""FUSE buffer vector structure."""
class fuse_operations(ctypes.Structure):
"""FUSE operations structure."""
class c_flock_t(ctypes.Structure):
"""File locking structure (platform-specific)."""Type Definitions
from typing import Union, Iterable
import ctypes
FieldsEntry = Union[tuple[str, type], tuple[str, type, int]]
ReadDirResult = Iterable[Union[str, tuple[str, dict[str, int], int], tuple[str, int, int]]]
FuseConfigPointer = ctypes.POINTER(fuse_config)
FuseConnInfoPointer = ctypes.POINTER(fuse_conn_info)Install with Tessl CLI
npx tessl i tessl/pypi-mfusepy