tessl/pypi-mfusepy

Ctypes bindings for the high-level API in libfuse 2 and 3

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-mfusepy
Author: tessl

Error handling mechanisms and logging utilities for FUSE filesystem implementations. These provide structured error reporting and debugging capabilities for filesystem operations.

Capabilities

FuseOSError Exception

The primary exception class for FUSE-related errors, providing errno-based error reporting that integrates with the FUSE kernel interface.

class FuseOSError(OSError):
    """Exception class for FUSE-related OS errors."""
    
    def __init__(self, errno):
        """
        Create a FUSE OS error.
        
        Args:
            errno: Error number (from errno module constants)
        """

Usage Examples

import errno
from mfusepy import FuseOSError, Operations

class MyFilesystem(Operations):
    def getattr(self, path, fh=None):
        if path not in self.files:
            # File not found
            raise FuseOSError(errno.ENOENT)
        
        if not self._has_permission(path, 'read'):
            # Permission denied
            raise FuseOSError(errno.EACCES)
        
        return self.files[path]
    
    def mkdir(self, path, mode):
        if path in self.files:
            # File already exists
            raise FuseOSError(errno.EEXIST)
        
        parent = os.path.dirname(path)
        if parent not in self.files:
            # Parent directory doesn't exist
            raise FuseOSError(errno.ENOENT)
        
        # Create directory
        self.files[path] = {'st_mode': mode | 0x4000, 'st_nlink': 2}
        return 0

Common Error Codes

import errno

# File system errors commonly used with FuseOSError:
errno.ENOENT    # No such file or directory
errno.EACCES    # Permission denied
errno.EEXIST    # File exists
errno.EISDIR    # Is a directory
errno.ENOTDIR   # Not a directory
errno.ENOTEMPTY # Directory not empty
errno.ENOSPC    # No space left on device
errno.EROFS     # Read-only file system
errno.EIO       # I/O error
errno.EFBIG     # File too large
errno.ENOSYS    # Function not implemented
errno.EINVAL    # Invalid argument

LoggingMixIn

A mixin class that automatically logs all filesystem operation calls and their results, useful for debugging and monitoring filesystem implementations.

class LoggingMixIn:
    """Mixin class for logging all Operation callbacks."""

Usage Example

from mfusepy import Operations, LoggingMixIn, FUSE
import logging

# Configure logging
logging.basicConfig(level=logging.DEBUG)

class MyFS(LoggingMixIn, Operations):
    """Filesystem with automatic logging."""
    
    def __init__(self):
        self.files = {'/': {'st_mode': 0o755 | 0x4000, 'st_nlink': 2}}
    
    def getattr(self, path, fh=None):
        if path in self.files:
            return self.files[path]
        raise FuseOSError(errno.ENOENT)
    
    def readdir(self, path, fh):
        return ['.', '..'] + [name[1:] for name in self.files if name != '/']

# Mount with logging - all operations will be automatically logged
filesystem = MyFS()
fuse = FUSE(filesystem, '/mnt/myfs', foreground=True)

# Log output will show:
# DEBUG:fuse.log-mixin:-> getattr / None
# DEBUG:fuse.log-mixin:<- getattr {'st_mode': 16877, 'st_nlink': 2}

Decorators

Utility decorators for method validation and logging in filesystem implementations.

from typing import Callable

def log_callback(method: Callable) -> Callable:
    """
    Decorator that adds log output for the decorated method.
    
    Args:
        method: Method to decorate
        
    Returns:
        Decorated method with logging
    """

def overrides(parent_class: type) -> Callable:
    """
    Decorator that checks method exists in parent class.
    
    Args:
        parent_class: Parent class to check against
        
    Returns:
        Decorator function
    """

Usage Examples

from mfusepy import Operations, log_callback, overrides

class MyFS(Operations):
    @overrides(Operations)
    @log_callback
    def getattr(self, path, fh=None):
        # Implementation with automatic validation and logging
        return {'st_mode': 0o644, 'st_nlink': 1, 'st_size': 0}
    
    @log_callback  
    def custom_method(self):
        # Custom method with logging
        return "custom result"

Error Handling Patterns

Basic Error Handling

from mfusepy import Operations, FuseOSError
import errno
import os

class RobustFS(Operations):
    def getattr(self, path, fh=None):
        try:
            # Attempt to get file attributes
            stat_result = self._get_file_stats(path)
            return {
                'st_mode': stat_result.mode,
                'st_size': stat_result.size,
                'st_nlink': stat_result.nlink,
            }
        except FileNotFoundError:
            raise FuseOSError(errno.ENOENT)
        except PermissionError:
            raise FuseOSError(errno.EACCES)
        except Exception as e:
            # Log unexpected errors and return generic I/O error
            self.logger.error(f"Unexpected error in getattr({path}): {e}")
            raise FuseOSError(errno.EIO)

Validation and Sanitization

class SecureFS(Operations):
    def _validate_path(self, path):
        """Validate and sanitize file paths."""
        if not path.startswith('/'):
            raise FuseOSError(errno.EINVAL)
        
        # Prevent directory traversal
        normalized = os.path.normpath(path)
        if '..' in normalized or normalized != path:
            raise FuseOSError(errno.EACCES)
        
        return normalized
    
    def _check_permissions(self, path, operation):
        """Check if operation is permitted on path."""
        if not self._has_permission(path, operation):
            raise FuseOSError(errno.EACCES)
    
    def read(self, path, size, offset, fh):
        path = self._validate_path(path)
        self._check_permissions(path, 'read')
        
        try:
            return self._read_file_data(path, size, offset)
        except IOError:
            raise FuseOSError(errno.EIO)

Resource Management

class ResourceFS(Operations):
    def __init__(self):
        self.open_files = {}
        self.max_open_files = 1000
    
    def open(self, path, flags):
        if len(self.open_files) >= self.max_open_files:
            raise FuseOSError(errno.EMFILE)  # Too many open files
        
        try:
            fh = self._allocate_file_handle()
            self.open_files[fh] = self._open_file(path, flags)
            return fh
        except IOError:
            raise FuseOSError(errno.EIO)
    
    def release(self, path, fh):
        try:
            if fh in self.open_files:
                self.open_files[fh].close()
                del self.open_files[fh]
            return 0
        except Exception as e:
            self.logger.warning(f"Error closing file handle {fh}: {e}")
            return 0  # Don't propagate errors from release

Logging Configuration

import logging
from mfusepy import LoggingMixIn, Operations, FUSE

# Configure different log levels for different components
logging.getLogger('fuse').setLevel(logging.INFO)
logging.getLogger('fuse.log-mixin').setLevel(logging.DEBUG)

# Custom formatter for FUSE logs
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

handler = logging.StreamHandler()
handler.setFormatter(formatter)

logger = logging.getLogger('fuse')
logger.addHandler(handler)

class MyFS(LoggingMixIn, Operations):
    # Your filesystem implementation
    pass

Install with Tessl CLI