Python bindings for GNU parted library providing disk partition management capabilities
—
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.
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."""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.
"""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")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')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 partitionimport 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()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')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")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()PartedException - Base class for all pyparted exceptionsException - Python base exception classObject Creation:
CreateException - Object creation failuresAlignmentException - Alignment creation/operation failuresConstraintException - Constraint creation/solving failuresDevice/Disk Operations:
DeviceException - Device access/operation failuresDiskException - Disk/partition table failuresDiskLabelException - Disk label specific failuresIOException - Low-level I/O failuresPartition Operations:
PartitionException - Partition-specific failuresGeometryException - Geometry calculation/validation failuresFileSystemException - Filesystem operation failuresDiscovery/Identification:
UnknownDeviceException - Device identification failuresUnknownTypeException - Type identification failuresInstall with Tessl CLI
npx tessl i tessl/pypi-pyparted