CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-pyparted

Python bindings for GNU parted library providing disk partition management capabilities

Pending
Overview
Eval results
Files

exception-handling.mddocs/

Exception Handling

pyparted provides comprehensive exception handling with specific exception types for different error conditions. The exception system includes both high-level Python exceptions and low-level libparted exception handling.

Capabilities

Exception Classes

Specific exception types for different categories of errors.

# Core operation exceptions
class AlignmentException(Exception):
    """Raised when alignment operations fail."""

class ConstraintException(Exception):  
    """Raised when constraint operations fail or constraints cannot be satisfied."""

class CreateException(Exception):
    """Raised when object creation fails."""

class DeviceException(Exception):
    """Raised when device operations fail."""

class DiskException(Exception):
    """Raised when disk operations fail."""

class DiskLabelException(Exception):
    """Raised when disk label operations fail."""

class FileSystemException(Exception):
    """Raised when filesystem operations fail."""

class GeometryException(Exception):
    """Raised when geometry operations fail."""

class IOException(Exception):
    """Raised when I/O operations fail."""

class PartitionException(Exception):
    """Raised when partition operations fail."""

class TimerException(Exception):
    """Raised when timer operations fail."""

# Discovery exceptions
class UnknownDeviceException(Exception):
    """Raised when device cannot be identified."""

class UnknownTypeException(Exception):
    """Raised when type cannot be identified or is invalid."""

# General exceptions
class PartedException(Exception):
    """General pyparted exception base class."""

class NotNeededException(Exception):
    """Raised when operation is not needed."""

# Property access exceptions  
class ReadOnlyProperty(Exception):
    """Raised when attempting to write to read-only property."""

class WriteOnlyProperty(Exception):
    """Raised when attempting to read from write-only property."""

Exception Handler Functions

Functions for managing low-level libparted exception handling.

def register_exn_handler(handler_function) -> None:
    """
    Register custom exception handler for libparted exceptions.
    
    Args:
        handler_function: Function that takes (exception_type, message, options)
                         and returns resolution constant
    """

def clear_exn_handler() -> None:
    """
    Clear registered exception handler, reverting to default handling.
    """

Usage Examples

Basic Exception Handling

import parted

def safe_device_access(device_path):
    """Safely access device with proper exception handling."""
    try:
        device = parted.getDevice(device_path)
        print(f"Device: {device.model}")
        return device
        
    except parted.DeviceException as e:
        print(f"Device error: {e}")
        return None
    except parted.UnknownDeviceException as e:
        print(f"Unknown device: {e}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

# Test with valid and invalid paths
devices_to_test = ['/dev/sda', '/dev/nonexistent', '/invalid/path']
for device_path in devices_to_test:
    result = safe_device_access(device_path)
    print(f"{device_path}: {'Success' if result else 'Failed'}\n")

Comprehensive Disk Operations with Exception Handling

import parted

def safe_disk_operations(device_path):
    """Perform disk operations with comprehensive error handling."""
    try:
        # Get device
        device = parted.getDevice(device_path)
        print(f"Found device: {device.model}")
        
        # Read disk
        try:
            disk = parted.newDisk(device)
            print(f"Disk type: {disk.type}")
            
            # List partitions
            for partition in disk.partitions:
                try:
                    size = partition.getSize('GB')
                    print(f"Partition {partition.number}: {size:.1f} GB")
                    
                    # Check flags safely
                    if partition.isFlagAvailable(parted.PARTITION_BOOT):
                        boot_flag = partition.getFlag(parted.PARTITION_BOOT)
                        print(f"  Boot flag: {boot_flag}")
                        
                except parted.PartitionException as e:
                    print(f"  Partition error: {e}")
                    
        except parted.DiskException as e:
            print(f"Disk error: {e}")
        except parted.DiskLabelException as e:
            print(f"Disk label error: {e}")
            
    except parted.DeviceException as e:
        print(f"Device error: {e}")
    except parted.IOException as e:
        print(f"I/O error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

safe_disk_operations('/dev/sda')

Partition Creation with Error Recovery

import parted

def create_partition_with_recovery(device_path, size_gb):
    """Create partition with error recovery and fallback strategies."""
    try:
        device = parted.getDevice(device_path)
        disk = parted.newDisk(device)
        
        # Calculate partition geometry
        size_sectors = int(size_gb * (1024**3) / device.sectorSize)
        start_sector = 2048
        
        try:
            # Try optimal alignment first
            constraint = device.getOptimalAlignedConstraint()
            
            try:
                geometry = parted.Geometry(device, start=start_sector, length=size_sectors)
                
                # Check if geometry satisfies constraint
                if constraint.isSolution(geometry):
                    print("Using requested geometry")
                    final_geometry = geometry
                else:
                    print("Adjusting geometry to satisfy constraints")
                    final_geometry = constraint.solveNearest(geometry)
                
                # Create partition
                partition = parted.Partition(
                    disk=disk,
                    type=parted.PARTITION_NORMAL,
                    geometry=final_geometry
                )
                
                disk.addPartition(partition, constraint)
                disk.commit()
                
                print(f"Successfully created partition: {partition.path}")
                return True
                
            except parted.GeometryException as e:
                print(f"Geometry error: {e}")
                # Try with minimal constraint
                print("Retrying with minimal alignment...")
                constraint = device.getMinimalAlignedConstraint()
                # ... retry logic here
                
        except parted.ConstraintException as e:
            print(f"Constraint error: {e}")
            # Try with device-wide constraint
            print("Retrying with device constraint...")
            constraint = device.getConstraint()
            # ... retry logic here
            
    except parted.CreateException as e:
        print(f"Creation failed: {e}")
        return False
    except parted.PartitionException as e:
        print(f"Partition operation failed: {e}")
        return False
    except parted.DiskException as e:
        print(f"Disk operation failed: {e}")
        return False
    except Exception as e:
        print(f"Unexpected error: {e}")
        return False

create_partition_with_recovery('/dev/sdb', 10)  # 10GB partition

Custom Exception Handler

import parted
import sys

def interactive_exception_handler(exception_type, message, options):
    """Interactive exception handler for user decisions."""
    
    # Map exception types to readable names
    type_names = {
        parted.EXCEPTION_TYPE_INFORMATION: "INFO",
        parted.EXCEPTION_TYPE_WARNING: "WARNING", 
        parted.EXCEPTION_TYPE_ERROR: "ERROR",
        parted.EXCEPTION_TYPE_FATAL: "FATAL",
        parted.EXCEPTION_TYPE_BUG: "BUG",
        parted.EXCEPTION_TYPE_NO_FEATURE: "NO_FEATURE"
    }
    
    type_name = type_names.get(exception_type, f"UNKNOWN({exception_type})")
    print(f"\n{type_name}: {message}")
    
    # Handle based on available options
    if options == parted.EXCEPTION_OPT_OK_CANCEL:
        response = input("Continue? (o)k/(c)ancel: ").lower()
        return parted.EXCEPTION_RESOLVE_OK if response == 'o' else parted.EXCEPTION_RESOLVE_CANCEL
        
    elif options == parted.EXCEPTION_OPT_YES_NO:
        response = input("Continue? (y)es/(n)o: ").lower()
        return parted.EXCEPTION_RESOLVE_YES if response == 'y' else parted.EXCEPTION_RESOLVE_NO
        
    elif options == parted.EXCEPTION_OPT_YES_NO_CANCEL:
        response = input("Action? (y)es/(n)o/(c)ancel: ").lower()
        if response == 'y':
            return parted.EXCEPTION_RESOLVE_YES
        elif response == 'n':
            return parted.EXCEPTION_RESOLVE_NO
        else:
            return parted.EXCEPTION_RESOLVE_CANCEL
            
    elif options == parted.EXCEPTION_OPT_RETRY_CANCEL:
        response = input("Action? (r)etry/(c)ancel: ").lower()
        return parted.EXCEPTION_RESOLVE_RETRY if response == 'r' else parted.EXCEPTION_RESOLVE_CANCEL
        
    elif options == parted.EXCEPTION_OPT_RETRY_IGNORE_CANCEL:
        response = input("Action? (r)etry/(i)gnore/(c)ancel: ").lower()
        if response == 'r':
            return parted.EXCEPTION_RESOLVE_RETRY
        elif response == 'i':
            return parted.EXCEPTION_RESOLVE_IGNORE
        else:
            return parted.EXCEPTION_RESOLVE_CANCEL
            
    elif options == parted.EXCEPTION_OPT_IGNORE_CANCEL:
        response = input("Action? (i)gnore/(c)ancel: ").lower()
        return parted.EXCEPTION_RESOLVE_IGNORE if response == 'i' else parted.EXCEPTION_RESOLVE_CANCEL
    
    # Default to unhandled
    return parted.EXCEPTION_RESOLVE_UNHANDLED

# Register the handler
parted.register_exn_handler(interactive_exception_handler)

# Use in operations
try:
    device = parted.getDevice('/dev/sdb')
    # ... perform operations that might trigger exceptions
    print("Operations completed")
finally:
    # Clean up handler
    parted.clear_exn_handler()

Logging Exception Handler

import parted
import logging
import datetime

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger('pyparted')

def logging_exception_handler(exception_type, message, options):
    """Exception handler that logs all exceptions."""
    
    type_names = {
        parted.EXCEPTION_TYPE_INFORMATION: "INFO",
        parted.EXCEPTION_TYPE_WARNING: "WARNING",
        parted.EXCEPTION_TYPE_ERROR: "ERROR", 
        parted.EXCEPTION_TYPE_FATAL: "FATAL",
        parted.EXCEPTION_TYPE_BUG: "BUG",
        parted.EXCEPTION_TYPE_NO_FEATURE: "NO_FEATURE"
    }
    
    type_name = type_names.get(exception_type, f"UNKNOWN({exception_type})")
    
    # Log the exception
    if exception_type == parted.EXCEPTION_TYPE_INFORMATION:
        logger.info(f"libparted: {message}")
        return parted.EXCEPTION_RESOLVE_OK
    elif exception_type == parted.EXCEPTION_TYPE_WARNING:
        logger.warning(f"libparted: {message}")
        return parted.EXCEPTION_RESOLVE_OK
    elif exception_type == parted.EXCEPTION_TYPE_ERROR:
        logger.error(f"libparted: {message}")
        # For non-interactive use, generally continue with "fix" or "ok"
        return parted.EXCEPTION_RESOLVE_FIX
    elif exception_type == parted.EXCEPTION_TYPE_FATAL:
        logger.critical(f"libparted: {message}")
        return parted.EXCEPTION_RESOLVE_CANCEL
    else:
        logger.error(f"libparted {type_name}: {message}")
        return parted.EXCEPTION_RESOLVE_UNHANDLED

# Register logging handler
parted.register_exn_handler(logging_exception_handler)

def logged_disk_operations(device_path):
    """Perform disk operations with comprehensive logging."""
    logger.info(f"Starting disk operations on {device_path}")
    
    try:
        device = parted.getDevice(device_path)
        logger.info(f"Device opened: {device.model}")
        
        disk = parted.newDisk(device)
        logger.info(f"Disk read: {disk.type} with {len(disk.partitions)} partitions")
        
        # Operations that might trigger libparted exceptions
        if not disk.check():
            logger.warning("Disk consistency check failed")
        
        return True
        
    except Exception as e:
        logger.error(f"Operation failed: {e}")
        return False
    finally:
        logger.info("Disk operations completed")

logged_disk_operations('/dev/sda')

Exception Context Manager

import parted
from contextlib import contextmanager

@contextmanager
def exception_handler_context(handler_func):
    """Context manager for temporary exception handler."""
    parted.register_exn_handler(handler_func)
    try:
        yield
    finally:
        parted.clear_exn_handler()

def silent_handler(exception_type, message, options):
    """Silent handler that automatically resolves exceptions."""
    if exception_type in [parted.EXCEPTION_TYPE_INFORMATION, parted.EXCEPTION_TYPE_WARNING]:
        return parted.EXCEPTION_RESOLVE_OK
    elif exception_type == parted.EXCEPTION_TYPE_ERROR:
        return parted.EXCEPTION_RESOLVE_FIX
    else:
        return parted.EXCEPTION_RESOLVE_CANCEL

# Use context manager for specific operations
with exception_handler_context(silent_handler):
    try:
        device = parted.getDevice('/dev/sda')
        disk = parted.newDisk(device)
        # Perform operations with silent exception handling
        print(f"Processed disk with {len(disk.partitions)} partitions")
    except Exception as e:
        print(f"Error: {e}")

print("Back to default exception handling")

Exception Analysis and Reporting

import parted
from collections import defaultdict

class ExceptionTracker:
    """Track and analyze exceptions during operations."""
    
    def __init__(self):
        self.exceptions = []
        self.counts = defaultdict(int)
    
    def handler(self, exception_type, message, options):
        """Exception handler that tracks exceptions."""
        self.exceptions.append({
            'type': exception_type,
            'message': message,
            'options': options,
            'timestamp': parted.time.time()
        })
        self.counts[exception_type] += 1
        
        # Provide reasonable default responses
        if exception_type in [parted.EXCEPTION_TYPE_INFORMATION, parted.EXCEPTION_TYPE_WARNING]:
            return parted.EXCEPTION_RESOLVE_OK
        elif exception_type == parted.EXCEPTION_TYPE_ERROR:
            return parted.EXCEPTION_RESOLVE_FIX
        else:
            return parted.EXCEPTION_RESOLVE_CANCEL
    
    def report(self):
        """Generate exception report."""
        if not self.exceptions:
            print("No exceptions occurred")
            return
        
        type_names = {
            parted.EXCEPTION_TYPE_INFORMATION: "INFO",
            parted.EXCEPTION_TYPE_WARNING: "WARNING",
            parted.EXCEPTION_TYPE_ERROR: "ERROR",
            parted.EXCEPTION_TYPE_FATAL: "FATAL",
            parted.EXCEPTION_TYPE_BUG: "BUG",
            parted.EXCEPTION_TYPE_NO_FEATURE: "NO_FEATURE"
        }
        
        print(f"\nException Report: {len(self.exceptions)} total exceptions")
        print("Summary by type:")
        for exc_type, count in self.counts.items():
            type_name = type_names.get(exc_type, f"UNKNOWN({exc_type})")
            print(f"  {type_name}: {count}")
        
        print("\nDetailed exceptions:")
        for i, exc in enumerate(self.exceptions, 1):
            type_name = type_names.get(exc['type'], f"UNKNOWN({exc['type']})")
            print(f"  {i}. {type_name}: {exc['message']}")

# Use exception tracker
tracker = ExceptionTracker()

with exception_handler_context(tracker.handler):
    try:
        # Perform operations that might generate exceptions
        devices = parted.getAllDevices()
        for device in devices:
            try:
                disk = parted.newDisk(device)
                disk.check()
            except:
                continue  # Continue processing other devices
    except Exception as e:
        print(f"Error: {e}")

tracker.report()

Exception Hierarchy

Base Exceptions

  • PartedException - Base class for all pyparted exceptions
  • Exception - Python base exception class

Specific Exception Categories

Object Creation:

  • CreateException - Object creation failures
  • AlignmentException - Alignment creation/operation failures
  • ConstraintException - Constraint creation/solving failures

Device/Disk Operations:

  • DeviceException - Device access/operation failures
  • DiskException - Disk/partition table failures
  • DiskLabelException - Disk label specific failures
  • IOException - Low-level I/O failures

Partition Operations:

  • PartitionException - Partition-specific failures
  • GeometryException - Geometry calculation/validation failures
  • FileSystemException - Filesystem operation failures

Discovery/Identification:

  • UnknownDeviceException - Device identification failures
  • UnknownTypeException - Type identification failures

Best Practices

  1. Specific Exception Handling: Catch specific exceptions rather than generic Exception
  2. Progressive Fallback: Try optimal approaches first, fall back to simpler ones
  3. Resource Cleanup: Use try/finally or context managers for cleanup
  4. User Feedback: Provide meaningful error messages to users
  5. Logging: Log exceptions for debugging and monitoring
  6. Exception Handlers: Use custom handlers for automated or interactive scenarios

Install with Tessl CLI

npx tessl i tessl/pypi-pyparted

docs

constants-reference.md

constraint-solving.md

device-management.md

disk-operations.md

exception-handling.md

filesystem-operations.md

geometry-alignment.md

index.md

partition-management.md

utility-functions.md

tile.json