LVM

This section documents the Logical Volume Management (LVM) functionality.

sts.lvm

LVM device management.

This package provides functionality for managing LVM devices: - Physical Volume (PV) operations - Volume Group (VG) operations - Logical Volume (LV) operations

LVM (Logical Volume Management) provides flexible disk space management: 1. Physical Volumes (PVs): Physical disks or partitions 2. Volume Groups (VGs): Pool of space from PVs 3. Logical Volumes (LVs): Virtual partitions from VG space

Key benefits: - Resize filesystems online - Snapshot and mirror volumes - Stripe across multiple disks - Move data between disks

LogicalVolume dataclass

Bases: LvmDevice

Logical Volume device.

A Logical Volume (LV) is a virtual partition created from VG space. LVs appear as block devices that can be formatted and mounted.

Key features: - Flexible sizing - Online resizing - Snapshots - Striping and mirroring - Thin provisioning

Parameters:

Name	Type	Description	Default
name	str | None	Device name (optional)	None
path	Path | str | None	Device path (optional, defaults to /dev//)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
model	str | None	Device model (optional)	None
yes	bool	Automatically answer yes to prompts	True
force	bool	Force operations without confirmation	False
vg	str | None	Volume group name (optional, discovered from device)	None
report	LVReport | None	LV report instance (optional, created automatically)	None
prevent_report_updates	bool	Prevent automatic report updates (defaults to False)	False

The LogicalVolume class now includes integrated report functionality: - Automatic report creation when name and vg are provided - Automatic report refresh after state-changing operations - Access to detailed LV information directly via report attributes - Prevention of updates via prevent_report_updates flag

Example

# Basic usage with automatic report
lv = LogicalVolume(name='lv0', vg='vg0')
lv.create(size='100M')
print(lv.report.lv_size)

# Prevent automatic updates
lv = LogicalVolume(name='lv0', vg='vg0', prevent_report_updates=True)

# Manual report refresh
lv.refresh_report()

Source code in sts_libs/src/sts/lvm/logical_volume.py

@dataclass
class LogicalVolume(LvmDevice):
    """Logical Volume device.

    A Logical Volume (LV) is a virtual partition created from VG space.
    LVs appear as block devices that can be formatted and mounted.

    Key features:
    - Flexible sizing
    - Online resizing
    - Snapshots
    - Striping and mirroring
    - Thin provisioning

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<vg>/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        vg: Volume group name (optional, discovered from device)
        report: LV report instance (optional, created automatically)
        prevent_report_updates: Prevent automatic report updates (defaults to False)

    The LogicalVolume class now includes integrated report functionality:
    - Automatic report creation when name and vg are provided
    - Automatic report refresh after state-changing operations
    - Access to detailed LV information directly via report attributes
    - Prevention of updates via prevent_report_updates flag

    Example:
        ```python
        # Basic usage with automatic report
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.create(size='100M')
        print(lv.report.lv_size)

        # Prevent automatic updates
        lv = LogicalVolume(name='lv0', vg='vg0', prevent_report_updates=True)

        # Manual report refresh
        lv.refresh_report()
        ```
    """

    # Optional parameters for this class
    vg: str | None = None  # Parent VG
    pool_name: str | None = None
    report: LVReport | None = field(default=None, repr=False)
    prevent_report_updates: bool = False

    # Available LV commands
    COMMANDS: ClassVar[list[str]] = [
        'lvchange',  # Change LV attributes
        'lvcreate',  # Create LV
        'lvconvert',  # Convert LV type
        'lvdisplay',  # Show LV details
        'lvextend',  # Increase LV size
        'lvreduce',  # Reduce LV size
        'lvremove',  # Remove LV
        'lvrename',  # Rename LV
        'lvresize',  # Change LV size
        'lvs',  # List LVs
        'lvscan',  # Scan for LVs
    ]

    def __post_init__(self) -> None:
        """Initialize Logical Volume.

        - Sets device path from name and VG
        - Discovers VG membership
        - Creates and updates from report
        """
        # Set path based on name and vg if not provided
        if not self.path and self.name and self.vg:
            self.path = f'/dev/{self.vg}/{self.name}'

        # Initialize parent class
        super().__post_init__()

    def refresh_report(self) -> bool:
        """Refresh LV report data.

        Creates or updates the LV report with the latest information.

        Returns:
            bool: True if refresh was successful
        """
        # Create new report if needed
        if not self.report:
            # Do not provide name and vg during init to prevent update
            self.report = LVReport()
            self.report.name = self.name
            self.report.vg = self.vg

        # Refresh the report data
        return self.report.refresh()

    def discover_vg(self) -> str | None:
        """Discover VG if name is available."""
        if self.name and not self.vg:
            result = run(f'lvs {self.name} -o vg_name --noheadings')
            if result.succeeded:
                self.vg = result.stdout.strip()
                return self.vg
        return None

    def create(self, *args: str, **options: str) -> bool:
        """Create Logical Volume.

        Creates a new LV in the specified VG:
        - Allocates space from VG
        - Creates device mapper device
        - Initializes LV metadata

        If pool_name is set, automatically adds --thinpool or --vdopool option
        depending on the volume type:
        - For VDO volumes (type='vdo'): uses --vdopool
        - For thin volumes: uses --thinpool

        Args:
            *args: Additional arguments passed to lvcreate
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Create regular LV
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.create(size='1G')
            True

            # Create thin volume (with pool_name set)
            thin_lv = LogicalVolume(name='thin1', vg='vg0', pool_name='pool1')
            thin_lv.create(virtualsize='500M')
            True

            # Create VDO volume (with pool_name set)
            vdo_lv = LogicalVolume(name='vdo1', vg='vg0', pool_name='vdopool1')
            vdo_lv.create(type='vdo', size='8G')
            True
            ```
        """
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.vg:
            logging.error('Volume group required')
            return False

        # If pool_name is set, automatically add the appropriate pool option
        # Use --vdopool for VDO volumes, --thinpool for thin volumes
        if self.pool_name:
            lv_type = options.get('type', '')
            if lv_type == 'vdo' and 'vdopool' not in options:
                options = {**options, 'vdopool': self.pool_name}
            elif 'thinpool' not in options and lv_type != 'vdo':
                options = {**options, 'thinpool': self.pool_name}

        result = self._run('lvcreate', '-n', self.name, self.vg, *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def remove(self, *args: str, **options: str) -> bool:
        """Remove Logical Volume.

        Removes LV and its data:
        - Data is permanently lost
        - Space is returned to VG
        - Device mapper device is removed

        Args:
            *args: Additional volume paths to remove (for removing multiple volumes)
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Remove single volume
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.remove()
            True

            # Remove multiple volumes
            lv = LogicalVolume(name='lv1', vg='vg0')
            lv.remove('vg0/lv2', 'vg0/lv3')
            True
            ```
        """
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.vg:
            logging.error('Volume group required')
            return False

        # Start with this LV
        targets = [f'{self.vg}/{self.name}']

        # Add any additional volumes from args
        if args:
            targets.extend(args)

        result = self._run('lvremove', *targets, **options)
        return result.succeeded

    def change(self, *args: str, **options: str) -> bool:
        """Change Logical Volume attributes.

        Change a general LV attribute:

        Args:
            *args: LV options (see LVMOptions)
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.change('-an', 'vg0/lv0')
            True
            ```
        """
        result = self._run('lvchange', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def extend(self, *args: str, **options: str) -> bool:
        """Extend Logical volume.

        - LV must be initialized (using lvcreate)
        - VG must have sufficient usable space

        Args:
            *args: Additional arguments passed to lvextend
            **options: All options passed as strings to lvextend command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lvol0', vg='vg0')
            lv.extend(extents='100%vg')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvextend', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def lvs(self, *args: str, **options: str) -> CommandResult:
        """Get information about logical volumes.

        Executes the 'lvs' command with optional filtering to display
        information about logical volumes.

        Args:
            *args: Positional args passed through to `lvs` (e.g., LV selector, flags).
            **options: LV command options (see LvmOptions).

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume()
            result = lv.lvs()
            print(result.stdout)
            ```
        """
        return self._run('lvs', *args, **options)

    def convert(self, *args: str, **options: str) -> bool:
        """Convert Logical Volume type.

        Converts LV type (linear, striped, mirror, snapshot, etc):
        - Can change between different LV types
        - May require additional space or devices
        - Some conversions are irreversible

        Args:
            *args: LV conversion arguments
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.convert('--type', 'mirror')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_to_thinpool(self, *args: str, **options: str) -> ThinPool | None:
        """Convert logical volume to thin pool.

        Converts an existing LV to a thin pool using lvconvert --thinpool.
        The LV must already exist and have sufficient space.

        Args:
            *args: Additional arguments passed to lvconvert
            **options: All options passed as strings to lvconvert command

        Returns:
            ThinPool instance if successful, None otherwise

        Example:
            ```python
            # Basic conversion
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.create(size='100M')
            pool = lv.convert_to_thinpool()

            # Conversion with parameters
            pool = lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

            # Conversion with separate metadata LV
            pool = lv.convert_to_thinpool(poolmetadata='metadata_lv')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return None
        if not self.name:
            logging.error('Logical volume name required')
            return None

        # Build the lvconvert command - pass all args and options through
        cmd_args = ['--thinpool', f'{self.vg}/{self.name}']
        if args:
            cmd_args.extend(args)

        result = self._run('lvconvert', *cmd_args, **options)
        if not result.succeeded:
            return None

        # Create and return new ThinPool instance based on current LV's properties
        thin_pool = ThinPool(
            name=self.name,
            vg=self.vg,
            yes=self.yes,
            force=self.force,
        )
        thin_pool.refresh_report()
        thin_pool.discover_thin_volumes()
        return thin_pool

    def convert_splitmirrors(self, *args: str, **options: str) -> tuple[bool, LogicalVolume | None]:
        """Split images from a raid1 or mirror LV and create a new LV.

        Splits the specified number of images from a raid1 or mirror LV and uses them
        to create a new LV with the specified name.

        Args:
            *args: All arguments passed to lvconvert (including count, new_name, etc.)
            **options: All options passed as strings to lvconvert command

        Returns:
            Tuple of (success, new_lv) where:
            - success: True if successful, False otherwise
            - new_lv: LogicalVolume object for the new split LV, or None if failed

        Example:
            ```python
            lv = LogicalVolume(name='mirror_lv', vg='vg0')
            success, split_lv = lv.convert_splitmirrors('--splitmirrors', '1', '--name', 'split_lv')
            if success:
                print(f'Created new LV: {split_lv.name}')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False, None
        if not self.name:
            logging.error('Logical volume name required')
            return False, None

        result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
        success = result.succeeded

        if success:
            self.refresh_report()

        # Try to find the new LV name from args (this is a best effort)
        new_lv = None
        if success:
            # Look for --name parameter in args
            new_name = None
            for i, arg in enumerate(args):
                if arg == '--name' and i + 1 < len(args):
                    new_name = args[i + 1]
                    break

            if new_name:
                try:
                    new_lv = LogicalVolume(name=new_name, vg=self.vg)
                    if not new_lv.refresh_report():
                        logging.warning(f'Failed to refresh report for new LV {new_name}')
                except (ValueError, OSError) as e:
                    logging.warning(f'Failed to create LogicalVolume object for {new_name}: {e}')
                    new_lv = None

        return success, new_lv

    def convert_originname(self, *args: str, **options: str) -> tuple[bool, LogicalVolume | None]:
        """Convert LV to thin LV with named external origin.

        Converts the LV to a thin LV in the specified thin pool, using the original LV
        as an external read-only origin with the specified name.

        Args:
            *args: Additional arguments passed to lvconvert (optional)
            **options: Options for the conversion:
                - thinpool: Name of the thin pool (required, format: vg/pool or pool)
                - originname: Name for the read-only origin LV (required)
                - type: LV type (defaults to 'thin')
                - Other lvconvert options as needed

        Returns:
            Tuple of (success, origin_lv) where:
            - success: True if successful, False otherwise
            - origin_lv: LogicalVolume object for the read-only origin LV, or None if failed

        Example:
            ```python
            # Simple usage with keyword arguments
            lv = LogicalVolume(name='data_lv', vg='vg0')
            success, origin_lv = lv.convert_originname(thinpool='vg0/thin_pool', originname='data_origin')

            # With additional options
            success, origin_lv = lv.convert_originname(
                thinpool='vg0/thin_pool', originname='data_origin', chunksize='128k'
            )
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False, None
        if not self.name:
            logging.error('Logical volume name required')
            return False, None

        # Extract required parameters
        thinpool = options.pop('thinpool', None)
        originname = options.pop('originname', None)

        if not thinpool:
            logging.error('thinpool parameter is required')
            return False, None
        if not originname:
            logging.error('originname parameter is required')
            return False, None

        # Build command arguments
        cmd_args = ['--type', options.pop('type', 'thin')]
        cmd_args.extend(['--thinpool', thinpool])
        cmd_args.extend(['--originname', originname])

        # Add any additional args passed positionally
        if args:
            cmd_args.extend(args)

        # Add any remaining options as --key value pairs
        for key, value in options.items():
            cmd_args.extend([f'--{key}', value])

        # Execute the conversion
        result = self._run('lvconvert', f'{self.vg}/{self.name}', *cmd_args)
        success = result.succeeded

        if success:
            self.refresh_report()

        # Create LogicalVolume object for the origin
        origin_lv = None
        if success and originname:
            try:
                origin_lv = LogicalVolume(name=originname, vg=self.vg)
                if not origin_lv.refresh_report():
                    logging.warning(f'Failed to refresh report for origin LV {originname}')
            except (ValueError, OSError) as e:
                logging.warning(f'Failed to create LogicalVolume object for {originname}: {e}')
                origin_lv = None

        return success, origin_lv

    def display(self, *args: str, **options: str) -> CommandResult:
        """Display Logical Volume details.

        Shows detailed information about the LV:
        - Size and allocation
        - Attributes and permissions
        - Segment information
        - Device mapper details

        Args:
            *args: Additional arguments passed to lvdisplay
            **options: All options passed as strings to lvdisplay command

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            result = lv.display()
            print(result.stdout)
            ```
        """
        if not self.vg or not self.name:
            return self._run('lvdisplay', *args, **options)
        return self._run('lvdisplay', f'{self.vg}/{self.name}', *args, **options)

    def reduce(self, *args: str, **options: str) -> bool:
        """Reduce Logical Volume size.

        Reduces LV size (shrinks the volume):
        - Filesystem must be shrunk first
        - Data loss risk if not done carefully
        - Cannot reduce below used space

        Args:
            *args: Additional lvreduce arguments
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.reduce(size='500M')
            True

            # With additional arguments
            lv.reduce('--test', size='500M')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        result = self._run('lvreduce', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def rename(self, new_name: str, *args: str, **options: str) -> bool:
        """Rename Logical Volume.

        Changes the LV name:
        - Must not conflict with existing LV names
        - Updates device mapper devices
        - May require remounting if mounted

        Args:
            new_name: New name for the LV
            *args: Additional arguments passed to lvrename
            **options: All options passed as strings to lvrename command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.rename('new_lv')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not new_name:
            logging.error('New name required')
            return False

        result = self._run('lvrename', f'{self.vg}/{self.name}', new_name, *args, **options)
        if result.succeeded:
            self.name = new_name
            self.path = f'/dev/{self.vg}/{self.name}'
            self.refresh_report()
        return result.succeeded

    def resize(self, *args: str, **options: str) -> bool:
        """Resize Logical Volume.

        Changes LV size (can grow or shrink):
        - Combines extend and reduce functionality
        - Safer than lvreduce for shrinking
        - Can resize filesystem simultaneously

        Args:
            *args: Additional lvresize arguments (e.g., '-l+2', '-t', '--test')
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.resize(size='2G')
            True

            # With additional arguments
            lv.resize('-l+2', size='2G')
            True

            # With test flag
            lv.resize('--test', size='2G')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvresize', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def scan(self, *args: str, **options: str) -> CommandResult:
        """Scan for Logical Volumes.

        Scans all devices for LV information:
        - Discovers new LVs
        - Updates device mapper
        - Useful after system changes

        Args:
            **options: LV options (see LvmOptions)

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume()
            result = lv.scan()
            print(result.stdout)
            ```
        """
        return self._run('lvscan', *args, **options)

    def deactivate(self) -> bool:
        """Deactivate Logical Volume."""
        udevadm_settle()
        result = self.change('-an', f'{self.vg}/{self.name}')
        udevadm_settle()
        if result:
            return self.wait_for_lv_deactivation()
        return result

    def activate(self) -> bool:
        """Activate Logical Volume."""
        return self.change('-ay', f'{self.vg}/{self.name}')

    def wait_for_lv_deactivation(self, timeout: int = 30) -> bool:
        """Wait for logical volume to be fully deactivated.

        Args:
            timeout: Maximum wait time in seconds

        Returns:
            True if deactivated successfully, False if timeout
        """
        start_time = time.time()

        while time.time() - start_time < timeout:
            # Check LV status using lvs command
            self.refresh_report()
            logging.info(self.report.lv_active if self.report else None)
            if self.report and self.report.lv_active != 'active':
                # LV is inactive - also verify device node is gone
                if self.path is not None:
                    device_path = Path(self.path)
                    if not device_path.exists():
                        return True
                else:
                    return True  # If no path, consider it deactivated
            time.sleep(2)  # Poll every 2 seconds

        logging.warning(f'LV {self.vg}/{self.name} deactivation timed out after {timeout}s')
        return False

    def change_discards(self, *args: str, **options: str) -> bool:
        """Change discards setting for logical volume.

        Args:
            *args: Additional arguments passed to lvchange
            **options: All options passed as strings to lvchange command

        Returns:
            True if change succeeded
        """
        if not self.vg or not self.name:
            logging.error('Volume group and logical volume name required')
            return False

        return self.change(f'{self.vg}/{self.name}', *args, **options)

    def create_snapshot(self, snapshot_name: str, *args: str, **options: str) -> LogicalVolume | None:
        """Create snapshot of this LV.

        Creates a snapshot of the current LV (self) with the given snapshot name.
        For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

        Args:
            snapshot_name: Name for the new snapshot LV
            *args: Additional command-line arguments (e.g., '-K', '-k', '--test')
            **options: Snapshot options (see LvmOptions)

        Returns:
            LogicalVolume instance representing the created snapshot, or None if failed

        Example:
            ```python
            # Create origin LV and then snapshot it
            origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
            # ... create the origin LV ...
            # Create thin snapshot (no size needed)
            snap1 = origin_lv.create_snapshot('snap1')

            # Create COW snapshot with ignore activation skip
            snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')
            ```
        """
        if not self.name:
            logging.error('Origin LV name required')
            return None
        if not self.vg:
            logging.error('Volume group required')
            return None
        if not snapshot_name:
            logging.error('Snapshot name required')
            return None

        # Build snapshot command
        cmd_args = ['-s']

        # Add any additional arguments
        if args:
            cmd_args.extend(args)

        # Add origin (this LV)
        cmd_args.append(f'{self.vg}/{self.name}')

        # Add snapshot name
        cmd_args.extend(['-n', snapshot_name])

        result = self._run('lvcreate', *cmd_args, **options)
        if result.succeeded:
            self.refresh_report()
            # Return new LogicalVolume instance representing the snapshot
            snap = LogicalVolume(name=snapshot_name, vg=self.vg)
            snap.refresh_report()
            return snap
        return None

    @classmethod
    def from_report(cls, report: LVReport) -> LogicalVolume | None:
        """Create LogicalVolume from LVReport.

        Args:
            report: LV report data

        Returns:
            LogicalVolume instance or None if invalid

        Example:
            ```python
            lv = LogicalVolume.from_report(report)
            ```
        """
        if not report.name or not report.vg:
            return None

        # Create LogicalVolume with report already attached
        return cls(
            name=report.name,
            vg=report.vg,
            path=report.lv_path,
            report=report,  # Attach the report directly
            prevent_report_updates=True,  # Avoid double refresh since report is already fresh
        )

    @classmethod
    def get_all(cls, vg: str | None = None) -> list[LogicalVolume]:
        """Get all Logical Volumes.

        Args:
            vg: Optional volume group to filter by

        Returns:
            List of LogicalVolume instances

        Example:
            ```python
            LogicalVolume.get_all()
            [LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]
            ```
        """
        logical_volumes: list[LogicalVolume] = []

        # Get all reports
        reports = LVReport.get_all(vg)

        # Create LogicalVolumes from reports
        logical_volumes.extend(lv for report in reports if (lv := cls.from_report(report)))

        return logical_volumes

    def __eq__(self, other: object) -> bool:
        """Compare two LogicalVolume instances for equality.

        Two LogicalVolume instances are considered equal if they have the same:
        - name
        - volume group (vg)
        - pool_name (if both have a pool_name)

        For thin LVs that belong to a pool, the pool_name must also match.
        For regular LVs or when either LV has no pool_name, only name and vg are compared.

        Args:
            other: Object to compare with

        Returns:
            True if the LogicalVolume instances are equal, False otherwise

        Example:
            ```python
            lv1 = LogicalVolume(name='lv0', vg='vg0')
            lv2 = LogicalVolume(name='lv0', vg='vg0')
            lv3 = LogicalVolume(name='lv1', vg='vg0')

            assert lv1 == lv2  # Same name and vg
            assert lv1 != lv3  # Different name

            # For thin LVs with pools
            thin1 = LogicalVolume(name='thin1', vg='vg0')
            thin1.pool_name = 'pool'
            thin2 = LogicalVolume(name='thin1', vg='vg0')
            thin2.pool_name = 'pool'
            thin3 = LogicalVolume(name='thin1', vg='vg0')
            thin3.pool_name = 'other_pool'

            assert thin1 == thin2  # Same name, vg, and pool
            assert thin1 != thin3  # Different pool
            ```
        """
        if not isinstance(other, LogicalVolume):
            return False
        if self.pool_name is None or other.pool_name is None:
            return self.name == other.name and self.vg == other.vg
        return self.name == other.name and self.vg == other.vg and self.pool_name == other.pool_name

    def __hash__(self) -> int:
        """Generate hash for LogicalVolume instance.

        The hash is based on the combination of name, volume group (vg), and pool_name.
        This allows LogicalVolume instances to be used in sets and as dictionary keys.
        The hash is consistent with the equality comparison in __eq__.

        Returns:
            int: Hash value based on (name, vg, pool_name) tuple

        Example:
            ```python
            lv1 = LogicalVolume(name='lv0', vg='vg0')
            lv2 = LogicalVolume(name='lv0', vg='vg0')

            # Can be used in sets
            lv_set = {lv1, lv2}  # Only one instance since they're equal

            # Can be used as dictionary keys
            lv_dict = {lv1: 'some_value'}
            ```
        """
        return hash((self.name, self.vg, self.pool_name))

__eq__(other)

Compare two LogicalVolume instances for equality.

Two LogicalVolume instances are considered equal if they have the same: - name - volume group (vg) - pool_name (if both have a pool_name)

For thin LVs that belong to a pool, the pool_name must also match. For regular LVs or when either LV has no pool_name, only name and vg are compared.

Parameters:

Name	Type	Description	Default
other	object	Object to compare with	required

Returns:

Type	Description
bool	True if the LogicalVolume instances are equal, False otherwise

Example

lv1 = LogicalVolume(name='lv0', vg='vg0')
lv2 = LogicalVolume(name='lv0', vg='vg0')
lv3 = LogicalVolume(name='lv1', vg='vg0')

assert lv1 == lv2  # Same name and vg
assert lv1 != lv3  # Different name

# For thin LVs with pools
thin1 = LogicalVolume(name='thin1', vg='vg0')
thin1.pool_name = 'pool'
thin2 = LogicalVolume(name='thin1', vg='vg0')
thin2.pool_name = 'pool'
thin3 = LogicalVolume(name='thin1', vg='vg0')
thin3.pool_name = 'other_pool'

assert thin1 == thin2  # Same name, vg, and pool
assert thin1 != thin3  # Different pool

Source code in sts_libs/src/sts/lvm/logical_volume.py

def __eq__(self, other: object) -> bool:
    """Compare two LogicalVolume instances for equality.

    Two LogicalVolume instances are considered equal if they have the same:
    - name
    - volume group (vg)
    - pool_name (if both have a pool_name)

    For thin LVs that belong to a pool, the pool_name must also match.
    For regular LVs or when either LV has no pool_name, only name and vg are compared.

    Args:
        other: Object to compare with

    Returns:
        True if the LogicalVolume instances are equal, False otherwise

    Example:
        ```python
        lv1 = LogicalVolume(name='lv0', vg='vg0')
        lv2 = LogicalVolume(name='lv0', vg='vg0')
        lv3 = LogicalVolume(name='lv1', vg='vg0')

        assert lv1 == lv2  # Same name and vg
        assert lv1 != lv3  # Different name

        # For thin LVs with pools
        thin1 = LogicalVolume(name='thin1', vg='vg0')
        thin1.pool_name = 'pool'
        thin2 = LogicalVolume(name='thin1', vg='vg0')
        thin2.pool_name = 'pool'
        thin3 = LogicalVolume(name='thin1', vg='vg0')
        thin3.pool_name = 'other_pool'

        assert thin1 == thin2  # Same name, vg, and pool
        assert thin1 != thin3  # Different pool
        ```
    """
    if not isinstance(other, LogicalVolume):
        return False
    if self.pool_name is None or other.pool_name is None:
        return self.name == other.name and self.vg == other.vg
    return self.name == other.name and self.vg == other.vg and self.pool_name == other.pool_name

__hash__()

Generate hash for LogicalVolume instance.

The hash is based on the combination of name, volume group (vg), and pool_name. This allows LogicalVolume instances to be used in sets and as dictionary keys. The hash is consistent with the equality comparison in eq.

Returns:

Name	Type	Description
int	int	Hash value based on (name, vg, pool_name) tuple

Example

lv1 = LogicalVolume(name='lv0', vg='vg0')
lv2 = LogicalVolume(name='lv0', vg='vg0')

# Can be used in sets
lv_set = {lv1, lv2}  # Only one instance since they're equal

# Can be used as dictionary keys
lv_dict = {lv1: 'some_value'}

Source code in sts_libs/src/sts/lvm/logical_volume.py

def __hash__(self) -> int:
    """Generate hash for LogicalVolume instance.

    The hash is based on the combination of name, volume group (vg), and pool_name.
    This allows LogicalVolume instances to be used in sets and as dictionary keys.
    The hash is consistent with the equality comparison in __eq__.

    Returns:
        int: Hash value based on (name, vg, pool_name) tuple

    Example:
        ```python
        lv1 = LogicalVolume(name='lv0', vg='vg0')
        lv2 = LogicalVolume(name='lv0', vg='vg0')

        # Can be used in sets
        lv_set = {lv1, lv2}  # Only one instance since they're equal

        # Can be used as dictionary keys
        lv_dict = {lv1: 'some_value'}
        ```
    """
    return hash((self.name, self.vg, self.pool_name))

__post_init__()

Initialize Logical Volume.

• Sets device path from name and VG
• Discovers VG membership
• Creates and updates from report

Source code in sts_libs/src/sts/lvm/logical_volume.py

def __post_init__(self) -> None:
    """Initialize Logical Volume.

    - Sets device path from name and VG
    - Discovers VG membership
    - Creates and updates from report
    """
    # Set path based on name and vg if not provided
    if not self.path and self.name and self.vg:
        self.path = f'/dev/{self.vg}/{self.name}'

    # Initialize parent class
    super().__post_init__()

activate()

Activate Logical Volume.

Source code in sts_libs/src/sts/lvm/logical_volume.py

def activate(self) -> bool:
    """Activate Logical Volume."""
    return self.change('-ay', f'{self.vg}/{self.name}')

change(*args, **options)

Change Logical Volume attributes.

Change a general LV attribute:

Parameters:

Name	Type	Description	Default
*args	str	LV options (see LVMOptions)	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lv0', vg='vg0')
lv.change('-an', 'vg0/lv0')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def change(self, *args: str, **options: str) -> bool:
    """Change Logical Volume attributes.

    Change a general LV attribute:

    Args:
        *args: LV options (see LVMOptions)
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.change('-an', 'vg0/lv0')
        True
        ```
    """
    result = self._run('lvchange', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

change_discards(*args, **options)

Change discards setting for logical volume.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvchange	()
**options	str	All options passed as strings to lvchange command	{}

Returns:

Type	Description
bool	True if change succeeded

Source code in sts_libs/src/sts/lvm/logical_volume.py

def change_discards(self, *args: str, **options: str) -> bool:
    """Change discards setting for logical volume.

    Args:
        *args: Additional arguments passed to lvchange
        **options: All options passed as strings to lvchange command

    Returns:
        True if change succeeded
    """
    if not self.vg or not self.name:
        logging.error('Volume group and logical volume name required')
        return False

    return self.change(f'{self.vg}/{self.name}', *args, **options)

convert(*args, **options)

Convert Logical Volume type.

Converts LV type (linear, striped, mirror, snapshot, etc): - Can change between different LV types - May require additional space or devices - Some conversions are irreversible

Parameters:

Name	Type	Description	Default
*args	str	LV conversion arguments	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lv0', vg='vg0')
lv.convert('--type', 'mirror')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert(self, *args: str, **options: str) -> bool:
    """Convert Logical Volume type.

    Converts LV type (linear, striped, mirror, snapshot, etc):
    - Can change between different LV types
    - May require additional space or devices
    - Some conversions are irreversible

    Args:
        *args: LV conversion arguments
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.convert('--type', 'mirror')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

convert_originname(*args, **options)

Convert LV to thin LV with named external origin.

Converts the LV to a thin LV in the specified thin pool, using the original LV as an external read-only origin with the specified name.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvconvert (optional)	()
**options	str	Options for the conversion: - thinpool: Name of the thin pool (required, format: vg/pool or pool) - originname: Name for the read-only origin LV (required) - type: LV type (defaults to 'thin') - Other lvconvert options as needed	{}

Returns:

Type	Description
bool	Tuple of (success, origin_lv) where:
LogicalVolume | None	success: True if successful, False otherwise
tuple[bool, LogicalVolume | None]	origin_lv: LogicalVolume object for the read-only origin LV, or None if failed

Example

# Simple usage with keyword arguments
lv = LogicalVolume(name='data_lv', vg='vg0')
success, origin_lv = lv.convert_originname(thinpool='vg0/thin_pool', originname='data_origin')

# With additional options
success, origin_lv = lv.convert_originname(
    thinpool='vg0/thin_pool', originname='data_origin', chunksize='128k'
)

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert_originname(self, *args: str, **options: str) -> tuple[bool, LogicalVolume | None]:
    """Convert LV to thin LV with named external origin.

    Converts the LV to a thin LV in the specified thin pool, using the original LV
    as an external read-only origin with the specified name.

    Args:
        *args: Additional arguments passed to lvconvert (optional)
        **options: Options for the conversion:
            - thinpool: Name of the thin pool (required, format: vg/pool or pool)
            - originname: Name for the read-only origin LV (required)
            - type: LV type (defaults to 'thin')
            - Other lvconvert options as needed

    Returns:
        Tuple of (success, origin_lv) where:
        - success: True if successful, False otherwise
        - origin_lv: LogicalVolume object for the read-only origin LV, or None if failed

    Example:
        ```python
        # Simple usage with keyword arguments
        lv = LogicalVolume(name='data_lv', vg='vg0')
        success, origin_lv = lv.convert_originname(thinpool='vg0/thin_pool', originname='data_origin')

        # With additional options
        success, origin_lv = lv.convert_originname(
            thinpool='vg0/thin_pool', originname='data_origin', chunksize='128k'
        )
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False, None
    if not self.name:
        logging.error('Logical volume name required')
        return False, None

    # Extract required parameters
    thinpool = options.pop('thinpool', None)
    originname = options.pop('originname', None)

    if not thinpool:
        logging.error('thinpool parameter is required')
        return False, None
    if not originname:
        logging.error('originname parameter is required')
        return False, None

    # Build command arguments
    cmd_args = ['--type', options.pop('type', 'thin')]
    cmd_args.extend(['--thinpool', thinpool])
    cmd_args.extend(['--originname', originname])

    # Add any additional args passed positionally
    if args:
        cmd_args.extend(args)

    # Add any remaining options as --key value pairs
    for key, value in options.items():
        cmd_args.extend([f'--{key}', value])

    # Execute the conversion
    result = self._run('lvconvert', f'{self.vg}/{self.name}', *cmd_args)
    success = result.succeeded

    if success:
        self.refresh_report()

    # Create LogicalVolume object for the origin
    origin_lv = None
    if success and originname:
        try:
            origin_lv = LogicalVolume(name=originname, vg=self.vg)
            if not origin_lv.refresh_report():
                logging.warning(f'Failed to refresh report for origin LV {originname}')
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to create LogicalVolume object for {originname}: {e}')
            origin_lv = None

    return success, origin_lv

convert_splitmirrors(*args, **options)

Split images from a raid1 or mirror LV and create a new LV.

Splits the specified number of images from a raid1 or mirror LV and uses them to create a new LV with the specified name.

Parameters:

Name	Type	Description	Default
*args	str	All arguments passed to lvconvert (including count, new_name, etc.)	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
bool	Tuple of (success, new_lv) where:
LogicalVolume | None	success: True if successful, False otherwise
tuple[bool, LogicalVolume | None]	new_lv: LogicalVolume object for the new split LV, or None if failed

Example

lv = LogicalVolume(name='mirror_lv', vg='vg0')
success, split_lv = lv.convert_splitmirrors('--splitmirrors', '1', '--name', 'split_lv')
if success:
    print(f'Created new LV: {split_lv.name}')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert_splitmirrors(self, *args: str, **options: str) -> tuple[bool, LogicalVolume | None]:
    """Split images from a raid1 or mirror LV and create a new LV.

    Splits the specified number of images from a raid1 or mirror LV and uses them
    to create a new LV with the specified name.

    Args:
        *args: All arguments passed to lvconvert (including count, new_name, etc.)
        **options: All options passed as strings to lvconvert command

    Returns:
        Tuple of (success, new_lv) where:
        - success: True if successful, False otherwise
        - new_lv: LogicalVolume object for the new split LV, or None if failed

    Example:
        ```python
        lv = LogicalVolume(name='mirror_lv', vg='vg0')
        success, split_lv = lv.convert_splitmirrors('--splitmirrors', '1', '--name', 'split_lv')
        if success:
            print(f'Created new LV: {split_lv.name}')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False, None
    if not self.name:
        logging.error('Logical volume name required')
        return False, None

    result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
    success = result.succeeded

    if success:
        self.refresh_report()

    # Try to find the new LV name from args (this is a best effort)
    new_lv = None
    if success:
        # Look for --name parameter in args
        new_name = None
        for i, arg in enumerate(args):
            if arg == '--name' and i + 1 < len(args):
                new_name = args[i + 1]
                break

        if new_name:
            try:
                new_lv = LogicalVolume(name=new_name, vg=self.vg)
                if not new_lv.refresh_report():
                    logging.warning(f'Failed to refresh report for new LV {new_name}')
            except (ValueError, OSError) as e:
                logging.warning(f'Failed to create LogicalVolume object for {new_name}: {e}')
                new_lv = None

    return success, new_lv

convert_to_thinpool(*args, **options)

Convert logical volume to thin pool.

Converts an existing LV to a thin pool using lvconvert --thinpool. The LV must already exist and have sufficient space.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvconvert	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
ThinPool | None	ThinPool instance if successful, None otherwise

Example

# Basic conversion
lv = LogicalVolume(name='lv0', vg='vg0')
lv.create(size='100M')
pool = lv.convert_to_thinpool()

# Conversion with parameters
pool = lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

# Conversion with separate metadata LV
pool = lv.convert_to_thinpool(poolmetadata='metadata_lv')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert_to_thinpool(self, *args: str, **options: str) -> ThinPool | None:
    """Convert logical volume to thin pool.

    Converts an existing LV to a thin pool using lvconvert --thinpool.
    The LV must already exist and have sufficient space.

    Args:
        *args: Additional arguments passed to lvconvert
        **options: All options passed as strings to lvconvert command

    Returns:
        ThinPool instance if successful, None otherwise

    Example:
        ```python
        # Basic conversion
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.create(size='100M')
        pool = lv.convert_to_thinpool()

        # Conversion with parameters
        pool = lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

        # Conversion with separate metadata LV
        pool = lv.convert_to_thinpool(poolmetadata='metadata_lv')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return None
    if not self.name:
        logging.error('Logical volume name required')
        return None

    # Build the lvconvert command - pass all args and options through
    cmd_args = ['--thinpool', f'{self.vg}/{self.name}']
    if args:
        cmd_args.extend(args)

    result = self._run('lvconvert', *cmd_args, **options)
    if not result.succeeded:
        return None

    # Create and return new ThinPool instance based on current LV's properties
    thin_pool = ThinPool(
        name=self.name,
        vg=self.vg,
        yes=self.yes,
        force=self.force,
    )
    thin_pool.refresh_report()
    thin_pool.discover_thin_volumes()
    return thin_pool

create(*args, **options)

Create Logical Volume.

Creates a new LV in the specified VG: - Allocates space from VG - Creates device mapper device - Initializes LV metadata

If pool_name is set, automatically adds --thinpool or --vdopool option depending on the volume type: - For VDO volumes (type='vdo'): uses --vdopool - For thin volumes: uses --thinpool

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvcreate	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Create regular LV
lv = LogicalVolume(name='lv0', vg='vg0')
lv.create(size='1G')
True

# Create thin volume (with pool_name set)
thin_lv = LogicalVolume(name='thin1', vg='vg0', pool_name='pool1')
thin_lv.create(virtualsize='500M')
True

# Create VDO volume (with pool_name set)
vdo_lv = LogicalVolume(name='vdo1', vg='vg0', pool_name='vdopool1')
vdo_lv.create(type='vdo', size='8G')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def create(self, *args: str, **options: str) -> bool:
    """Create Logical Volume.

    Creates a new LV in the specified VG:
    - Allocates space from VG
    - Creates device mapper device
    - Initializes LV metadata

    If pool_name is set, automatically adds --thinpool or --vdopool option
    depending on the volume type:
    - For VDO volumes (type='vdo'): uses --vdopool
    - For thin volumes: uses --thinpool

    Args:
        *args: Additional arguments passed to lvcreate
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Create regular LV
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.create(size='1G')
        True

        # Create thin volume (with pool_name set)
        thin_lv = LogicalVolume(name='thin1', vg='vg0', pool_name='pool1')
        thin_lv.create(virtualsize='500M')
        True

        # Create VDO volume (with pool_name set)
        vdo_lv = LogicalVolume(name='vdo1', vg='vg0', pool_name='vdopool1')
        vdo_lv.create(type='vdo', size='8G')
        True
        ```
    """
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.vg:
        logging.error('Volume group required')
        return False

    # If pool_name is set, automatically add the appropriate pool option
    # Use --vdopool for VDO volumes, --thinpool for thin volumes
    if self.pool_name:
        lv_type = options.get('type', '')
        if lv_type == 'vdo' and 'vdopool' not in options:
            options = {**options, 'vdopool': self.pool_name}
        elif 'thinpool' not in options and lv_type != 'vdo':
            options = {**options, 'thinpool': self.pool_name}

    result = self._run('lvcreate', '-n', self.name, self.vg, *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

create_snapshot(snapshot_name, *args, **options)

Create snapshot of this LV.

Creates a snapshot of the current LV (self) with the given snapshot name. For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

Parameters:

Name	Type	Description	Default
snapshot_name	str	Name for the new snapshot LV	required
*args	str	Additional command-line arguments (e.g., '-K', '-k', '--test')	()
**options	str	Snapshot options (see LvmOptions)	{}

Returns:

Type	Description
LogicalVolume | None	LogicalVolume instance representing the created snapshot, or None if failed

Example

# Create origin LV and then snapshot it
origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
# ... create the origin LV ...
# Create thin snapshot (no size needed)
snap1 = origin_lv.create_snapshot('snap1')

# Create COW snapshot with ignore activation skip
snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def create_snapshot(self, snapshot_name: str, *args: str, **options: str) -> LogicalVolume | None:
    """Create snapshot of this LV.

    Creates a snapshot of the current LV (self) with the given snapshot name.
    For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

    Args:
        snapshot_name: Name for the new snapshot LV
        *args: Additional command-line arguments (e.g., '-K', '-k', '--test')
        **options: Snapshot options (see LvmOptions)

    Returns:
        LogicalVolume instance representing the created snapshot, or None if failed

    Example:
        ```python
        # Create origin LV and then snapshot it
        origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
        # ... create the origin LV ...
        # Create thin snapshot (no size needed)
        snap1 = origin_lv.create_snapshot('snap1')

        # Create COW snapshot with ignore activation skip
        snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')
        ```
    """
    if not self.name:
        logging.error('Origin LV name required')
        return None
    if not self.vg:
        logging.error('Volume group required')
        return None
    if not snapshot_name:
        logging.error('Snapshot name required')
        return None

    # Build snapshot command
    cmd_args = ['-s']

    # Add any additional arguments
    if args:
        cmd_args.extend(args)

    # Add origin (this LV)
    cmd_args.append(f'{self.vg}/{self.name}')

    # Add snapshot name
    cmd_args.extend(['-n', snapshot_name])

    result = self._run('lvcreate', *cmd_args, **options)
    if result.succeeded:
        self.refresh_report()
        # Return new LogicalVolume instance representing the snapshot
        snap = LogicalVolume(name=snapshot_name, vg=self.vg)
        snap.refresh_report()
        return snap
    return None

deactivate()

Deactivate Logical Volume.

Source code in sts_libs/src/sts/lvm/logical_volume.py

def deactivate(self) -> bool:
    """Deactivate Logical Volume."""
    udevadm_settle()
    result = self.change('-an', f'{self.vg}/{self.name}')
    udevadm_settle()
    if result:
        return self.wait_for_lv_deactivation()
    return result

discover_vg()

Discover VG if name is available.

Source code in sts_libs/src/sts/lvm/logical_volume.py

def discover_vg(self) -> str | None:
    """Discover VG if name is available."""
    if self.name and not self.vg:
        result = run(f'lvs {self.name} -o vg_name --noheadings')
        if result.succeeded:
            self.vg = result.stdout.strip()
            return self.vg
    return None

display(*args, **options)

Display Logical Volume details.

Shows detailed information about the LV: - Size and allocation - Attributes and permissions - Segment information - Device mapper details

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvdisplay	()
**options	str	All options passed as strings to lvdisplay command	{}

Returns:

Type	Description
CommandResult	CommandResult object containing command output and status

Example

lv = LogicalVolume(name='lv0', vg='vg0')
result = lv.display()
print(result.stdout)

Source code in sts_libs/src/sts/lvm/logical_volume.py

def display(self, *args: str, **options: str) -> CommandResult:
    """Display Logical Volume details.

    Shows detailed information about the LV:
    - Size and allocation
    - Attributes and permissions
    - Segment information
    - Device mapper details

    Args:
        *args: Additional arguments passed to lvdisplay
        **options: All options passed as strings to lvdisplay command

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        result = lv.display()
        print(result.stdout)
        ```
    """
    if not self.vg or not self.name:
        return self._run('lvdisplay', *args, **options)
    return self._run('lvdisplay', f'{self.vg}/{self.name}', *args, **options)

extend(*args, **options)

Extend Logical volume.

• LV must be initialized (using lvcreate)
• VG must have sufficient usable space

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvextend	()
**options	str	All options passed as strings to lvextend command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lvol0', vg='vg0')
lv.extend(extents='100%vg')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def extend(self, *args: str, **options: str) -> bool:
    """Extend Logical volume.

    - LV must be initialized (using lvcreate)
    - VG must have sufficient usable space

    Args:
        *args: Additional arguments passed to lvextend
        **options: All options passed as strings to lvextend command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lvol0', vg='vg0')
        lv.extend(extents='100%vg')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvextend', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

from_report(report) classmethod

Create LogicalVolume from LVReport.

Parameters:

Name	Type	Description	Default
report	LVReport	LV report data	required

Returns:

Type	Description
LogicalVolume | None	LogicalVolume instance or None if invalid

Example

lv = LogicalVolume.from_report(report)

Source code in sts_libs/src/sts/lvm/logical_volume.py

@classmethod
def from_report(cls, report: LVReport) -> LogicalVolume | None:
    """Create LogicalVolume from LVReport.

    Args:
        report: LV report data

    Returns:
        LogicalVolume instance or None if invalid

    Example:
        ```python
        lv = LogicalVolume.from_report(report)
        ```
    """
    if not report.name or not report.vg:
        return None

    # Create LogicalVolume with report already attached
    return cls(
        name=report.name,
        vg=report.vg,
        path=report.lv_path,
        report=report,  # Attach the report directly
        prevent_report_updates=True,  # Avoid double refresh since report is already fresh
    )

get_all(vg=None) classmethod

Get all Logical Volumes.

Parameters:

Name	Type	Description	Default
vg	str | None	Optional volume group to filter by	None

Returns:

Type	Description
list[LogicalVolume]	List of LogicalVolume instances

Example

LogicalVolume.get_all()
[LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]

Source code in sts_libs/src/sts/lvm/logical_volume.py

@classmethod
def get_all(cls, vg: str | None = None) -> list[LogicalVolume]:
    """Get all Logical Volumes.

    Args:
        vg: Optional volume group to filter by

    Returns:
        List of LogicalVolume instances

    Example:
        ```python
        LogicalVolume.get_all()
        [LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]
        ```
    """
    logical_volumes: list[LogicalVolume] = []

    # Get all reports
    reports = LVReport.get_all(vg)

    # Create LogicalVolumes from reports
    logical_volumes.extend(lv for report in reports if (lv := cls.from_report(report)))

    return logical_volumes

lvs(*args, **options)

Get information about logical volumes.

Executes the 'lvs' command with optional filtering to display information about logical volumes.

Parameters:

Name	Type	Description	Default
*args	str	Positional args passed through to lvs (e.g., LV selector, flags).	()
**options	str	LV command options (see LvmOptions).	{}

Returns:

Type	Description
CommandResult	CommandResult object containing command output and status

Example

lv = LogicalVolume()
result = lv.lvs()
print(result.stdout)

Source code in sts_libs/src/sts/lvm/logical_volume.py

def lvs(self, *args: str, **options: str) -> CommandResult:
    """Get information about logical volumes.

    Executes the 'lvs' command with optional filtering to display
    information about logical volumes.

    Args:
        *args: Positional args passed through to `lvs` (e.g., LV selector, flags).
        **options: LV command options (see LvmOptions).

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume()
        result = lv.lvs()
        print(result.stdout)
        ```
    """
    return self._run('lvs', *args, **options)

reduce(*args, **options)

Reduce Logical Volume size.

Reduces LV size (shrinks the volume): - Filesystem must be shrunk first - Data loss risk if not done carefully - Cannot reduce below used space

Parameters:

Name	Type	Description	Default
*args	str	Additional lvreduce arguments	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lv0', vg='vg0')
lv.reduce(size='500M')
True

# With additional arguments
lv.reduce('--test', size='500M')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def reduce(self, *args: str, **options: str) -> bool:
    """Reduce Logical Volume size.

    Reduces LV size (shrinks the volume):
    - Filesystem must be shrunk first
    - Data loss risk if not done carefully
    - Cannot reduce below used space

    Args:
        *args: Additional lvreduce arguments
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.reduce(size='500M')
        True

        # With additional arguments
        lv.reduce('--test', size='500M')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    result = self._run('lvreduce', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

refresh_report()

Refresh LV report data.

Creates or updates the LV report with the latest information.

Returns:

Name	Type	Description
bool	bool	True if refresh was successful

Source code in sts_libs/src/sts/lvm/logical_volume.py

def refresh_report(self) -> bool:
    """Refresh LV report data.

    Creates or updates the LV report with the latest information.

    Returns:
        bool: True if refresh was successful
    """
    # Create new report if needed
    if not self.report:
        # Do not provide name and vg during init to prevent update
        self.report = LVReport()
        self.report.name = self.name
        self.report.vg = self.vg

    # Refresh the report data
    return self.report.refresh()

remove(*args, **options)

Remove Logical Volume.

Removes LV and its data: - Data is permanently lost - Space is returned to VG - Device mapper device is removed

Parameters:

Name	Type	Description	Default
*args	str	Additional volume paths to remove (for removing multiple volumes)	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Remove single volume
lv = LogicalVolume(name='lv0', vg='vg0')
lv.remove()
True

# Remove multiple volumes
lv = LogicalVolume(name='lv1', vg='vg0')
lv.remove('vg0/lv2', 'vg0/lv3')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def remove(self, *args: str, **options: str) -> bool:
    """Remove Logical Volume.

    Removes LV and its data:
    - Data is permanently lost
    - Space is returned to VG
    - Device mapper device is removed

    Args:
        *args: Additional volume paths to remove (for removing multiple volumes)
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Remove single volume
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.remove()
        True

        # Remove multiple volumes
        lv = LogicalVolume(name='lv1', vg='vg0')
        lv.remove('vg0/lv2', 'vg0/lv3')
        True
        ```
    """
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.vg:
        logging.error('Volume group required')
        return False

    # Start with this LV
    targets = [f'{self.vg}/{self.name}']

    # Add any additional volumes from args
    if args:
        targets.extend(args)

    result = self._run('lvremove', *targets, **options)
    return result.succeeded

rename(new_name, *args, **options)

Rename Logical Volume.

Changes the LV name: - Must not conflict with existing LV names - Updates device mapper devices - May require remounting if mounted

Parameters:

Name	Type	Description	Default
new_name	str	New name for the LV	required
*args	str	Additional arguments passed to lvrename	()
**options	str	All options passed as strings to lvrename command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lv0', vg='vg0')
lv.rename('new_lv')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def rename(self, new_name: str, *args: str, **options: str) -> bool:
    """Rename Logical Volume.

    Changes the LV name:
    - Must not conflict with existing LV names
    - Updates device mapper devices
    - May require remounting if mounted

    Args:
        new_name: New name for the LV
        *args: Additional arguments passed to lvrename
        **options: All options passed as strings to lvrename command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.rename('new_lv')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not new_name:
        logging.error('New name required')
        return False

    result = self._run('lvrename', f'{self.vg}/{self.name}', new_name, *args, **options)
    if result.succeeded:
        self.name = new_name
        self.path = f'/dev/{self.vg}/{self.name}'
        self.refresh_report()
    return result.succeeded

resize(*args, **options)

Resize Logical Volume.

Changes LV size (can grow or shrink): - Combines extend and reduce functionality - Safer than lvreduce for shrinking - Can resize filesystem simultaneously

Parameters:

Name	Type	Description	Default
*args	str	Additional lvresize arguments (e.g., '-l+2', '-t', '--test')	()
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

lv = LogicalVolume(name='lv0', vg='vg0')
lv.resize(size='2G')
True

# With additional arguments
lv.resize('-l+2', size='2G')
True

# With test flag
lv.resize('--test', size='2G')
True

Source code in sts_libs/src/sts/lvm/logical_volume.py

def resize(self, *args: str, **options: str) -> bool:
    """Resize Logical Volume.

    Changes LV size (can grow or shrink):
    - Combines extend and reduce functionality
    - Safer than lvreduce for shrinking
    - Can resize filesystem simultaneously

    Args:
        *args: Additional lvresize arguments (e.g., '-l+2', '-t', '--test')
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.resize(size='2G')
        True

        # With additional arguments
        lv.resize('-l+2', size='2G')
        True

        # With test flag
        lv.resize('--test', size='2G')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvresize', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

scan(*args, **options)

Scan for Logical Volumes.

Scans all devices for LV information: - Discovers new LVs - Updates device mapper - Useful after system changes

Parameters:

Name	Type	Description	Default
**options	str	LV options (see LvmOptions)	{}

Returns:

Type	Description
CommandResult	CommandResult object containing command output and status

Example

lv = LogicalVolume()
result = lv.scan()
print(result.stdout)

Source code in sts_libs/src/sts/lvm/logical_volume.py

def scan(self, *args: str, **options: str) -> CommandResult:
    """Scan for Logical Volumes.

    Scans all devices for LV information:
    - Discovers new LVs
    - Updates device mapper
    - Useful after system changes

    Args:
        **options: LV options (see LvmOptions)

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume()
        result = lv.scan()
        print(result.stdout)
        ```
    """
    return self._run('lvscan', *args, **options)

wait_for_lv_deactivation(timeout=30)

Wait for logical volume to be fully deactivated.

Parameters:

Name	Type	Description	Default
timeout	int	Maximum wait time in seconds	30

Returns:

Type	Description
bool	True if deactivated successfully, False if timeout

Source code in sts_libs/src/sts/lvm/logical_volume.py

def wait_for_lv_deactivation(self, timeout: int = 30) -> bool:
    """Wait for logical volume to be fully deactivated.

    Args:
        timeout: Maximum wait time in seconds

    Returns:
        True if deactivated successfully, False if timeout
    """
    start_time = time.time()

    while time.time() - start_time < timeout:
        # Check LV status using lvs command
        self.refresh_report()
        logging.info(self.report.lv_active if self.report else None)
        if self.report and self.report.lv_active != 'active':
            # LV is inactive - also verify device node is gone
            if self.path is not None:
                device_path = Path(self.path)
                if not device_path.exists():
                    return True
            else:
                return True  # If no path, consider it deactivated
        time.sleep(2)  # Poll every 2 seconds

    logging.warning(f'LV {self.vg}/{self.name} deactivation timed out after {timeout}s')
    return False

LvmConfig dataclass

LVM configuration manager.

Provides methods to read and update LVM configuration values in /etc/lvm/lvm.conf.

Parameters:

Name	Type	Description	Default
config_path	Path	Path to lvm.conf (default: /etc/lvm/lvm.conf)	(lambda: Path('/etc/lvm/lvm.conf'))()

Key configuration options for thin provisioning: - thin_pool_autoextend_threshold: When to trigger autoextend (default: 100) - thin_pool_autoextend_percent: How much to extend (default: 20) - thin_pool_metadata_require_separate_pvs: Require separate PV for metadata (default: 0)

Note

LVM config files typically have default values commented out (e.g., # key = value). The get method reads both commented and uncommented values. The set method will uncomment a line when setting a value.

Example

# Create config manager
config = LvmConfig()

# Read a value (works for both commented and uncommented)
threshold = config.get('thin_pool_autoextend_threshold')
print(f'Autoextend threshold: {threshold}')

# Update a value (uncomments if necessary)
config.set('thin_pool_autoextend_threshold', '80')

# Update multiple values
config.set_multiple(
    {
        'thin_pool_autoextend_threshold': '80',
        'thin_pool_autoextend_percent': '50',
    }
)

Source code in sts_libs/src/sts/lvm/lvconf.py

@dataclass
class LvmConfig:
    """LVM configuration manager.

    Provides methods to read and update LVM configuration values
    in /etc/lvm/lvm.conf.

    Args:
        config_path: Path to lvm.conf (default: /etc/lvm/lvm.conf)

    Key configuration options for thin provisioning:
    - thin_pool_autoextend_threshold: When to trigger autoextend (default: 100)
    - thin_pool_autoextend_percent: How much to extend (default: 20)
    - thin_pool_metadata_require_separate_pvs: Require separate PV for metadata (default: 0)

    Note:
        LVM config files typically have default values commented out (e.g., `# key = value`).
        The `get` method reads both commented and uncommented values.
        The `set` method will uncomment a line when setting a value.

    Example:
        ```python
        # Create config manager
        config = LvmConfig()

        # Read a value (works for both commented and uncommented)
        threshold = config.get('thin_pool_autoextend_threshold')
        print(f'Autoextend threshold: {threshold}')

        # Update a value (uncomments if necessary)
        config.set('thin_pool_autoextend_threshold', '80')

        # Update multiple values
        config.set_multiple(
            {
                'thin_pool_autoextend_threshold': '80',
                'thin_pool_autoextend_percent': '50',
            }
        )
        ```
    """

    config_path: Path = field(default_factory=lambda: Path('/etc/lvm/lvm.conf'))

    # Common thin provisioning configuration keys
    THIN_POOL_AUTOEXTEND_THRESHOLD: str = 'thin_pool_autoextend_threshold'
    THIN_POOL_AUTOEXTEND_PERCENT: str = 'thin_pool_autoextend_percent'
    THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS: str = 'thin_pool_metadata_require_separate_pvs'

    def exists(self) -> bool:
        """Check if the configuration file exists.

        Returns:
            True if the configuration file exists, False otherwise
        """
        return self.config_path.exists()

    def get(self, key: str) -> str | None:
        """Get configuration value from lvm.conf.

        Searches for the specified key in lvm.conf and returns its value.
        Handles both commented (default) and uncommented (active) values.
        If both exist, returns the uncommented (active) value.

        Args:
            key: Configuration key to look up

        Returns:
            Configuration value if found, None otherwise

        Example:
            ```python
            config = LvmConfig()
            threshold = config.get('thin_pool_autoextend_threshold')
            # Returns '100' (default value, even if commented)
            ```
        """
        if not self.exists():
            logging.warning(f'LVM config file {self.config_path} not found')
            return None

        # Pattern to match both commented and uncommented key = value
        # Captures: (optional #) (whitespace) key (whitespace) = (whitespace) value
        # The value pattern handles the last occurrence of "key = value" format
        search_regex = re.compile(rf'^\s*#?\s*{re.escape(key)}\s*=\s*(\S+)')

        uncommented_value = None
        commented_value = None

        try:
            for line in self.config_path.read_text().splitlines():
                match = search_regex.match(line)
                if match:
                    value = match.group(1)
                    # Check if this is a commented line
                    if line.strip().startswith('#'):
                        commented_value = value
                    else:
                        uncommented_value = value
        except OSError:
            logging.exception(f'Failed to read LVM config file {self.config_path}')
            return None

        # Prefer uncommented (active) value over commented (default) value
        return uncommented_value if uncommented_value is not None else commented_value

    def set(self, key: str, value: str) -> bool:
        """Update a configuration value in lvm.conf.

        Searches for the specified key in lvm.conf and updates its value.
        If the key is commented out, it will be uncommented.
        Preserves the original indentation.

        Args:
            key: Configuration key to update
            value: New value to set

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            config = LvmConfig()
            config.set('thin_pool_autoextend_threshold', '80')
            ```
        """
        if not self.exists():
            logging.error(f'LVM config file {self.config_path} not found')
            return False

        # Pattern to match both commented and uncommented key = value
        # Captures leading whitespace and optional comment marker
        search_regex = re.compile(rf'^(\s*)(#?\s*){re.escape(key)}(\s*)=(\s*)\S*')

        try:
            lines = self.config_path.read_text().splitlines()
            updated_lines = []
            found = False

            for line in lines:
                match = search_regex.match(line)
                if match and not found:
                    # Get the leading whitespace (indentation)
                    indent = match.group(1)
                    # Create new line without comment marker
                    updated_lines.append(f'{indent}{key} = {value}')
                    found = True
                else:
                    updated_lines.append(line)

            if not found:
                logging.warning(f'Configuration key {key} not found in {self.config_path}')
                return False

            self.config_path.write_text('\n'.join(updated_lines) + '\n')

        except OSError:
            logging.exception(f'Failed to update LVM config file {self.config_path}')
            return False

        return True

    def set_multiple(self, settings: dict[str, str]) -> bool:
        """Update multiple configuration values in lvm.conf.

        Updates all specified key-value pairs in a single operation.
        More efficient than calling set() multiple times.
        If keys are commented out, they will be uncommented.

        Args:
            settings: Dictionary of key-value pairs to update

        Returns:
            True if all updates were successful, False otherwise

        Example:
            ```python
            config = LvmConfig()
            config.set_multiple(
                {
                    'thin_pool_autoextend_threshold': '80',
                    'thin_pool_autoextend_percent': '50',
                }
            )
            ```
        """
        if not self.exists():
            logging.error(f'LVM config file {self.config_path} not found')
            return False

        try:
            lines = self.config_path.read_text().splitlines()
            updated_lines = []
            keys_found: set[str] = set()

            for line in lines:
                updated_line = line
                for key, value in settings.items():
                    if key in keys_found:
                        continue
                    # Pattern to match both commented and uncommented key = value
                    search_regex = re.compile(rf'^(\s*)(#?\s*){re.escape(key)}(\s*)=(\s*)\S*')
                    match = search_regex.match(line)
                    if match:
                        # Get the leading whitespace (indentation)
                        indent = match.group(1)
                        # Create new line without comment marker
                        updated_line = f'{indent}{key} = {value}'
                        keys_found.add(key)
                        break
                updated_lines.append(updated_line)

            # Check if all keys were found
            missing_keys = set(settings.keys()) - keys_found
            if missing_keys:
                logging.warning(f'Configuration keys not found: {missing_keys}')
                return False

            self.config_path.write_text('\n'.join(updated_lines) + '\n')

        except OSError:
            logging.exception(f'Failed to update LVM config file {self.config_path}')
            return False

        return True

    def get_thin_pool_autoextend_threshold(self) -> str | None:
        """Get thin pool autoextend threshold.

        Returns:
            Threshold value as string, or None if not found
        """
        return self.get(self.THIN_POOL_AUTOEXTEND_THRESHOLD)

    def set_thin_pool_autoextend_threshold(self, value: str) -> bool:
        """Set thin pool autoextend threshold.

        Args:
            value: Threshold value (0-100)

        Returns:
            True if successful, False otherwise
        """
        return self.set(self.THIN_POOL_AUTOEXTEND_THRESHOLD, value)

    def get_thin_pool_autoextend_percent(self) -> str | None:
        """Get thin pool autoextend percent.

        Returns:
            Percent value as string, or None if not found
        """
        return self.get(self.THIN_POOL_AUTOEXTEND_PERCENT)

    def set_thin_pool_autoextend_percent(self, value: str) -> bool:
        """Set thin pool autoextend percent.

        Args:
            value: Percent value

        Returns:
            True if successful, False otherwise
        """
        return self.set(self.THIN_POOL_AUTOEXTEND_PERCENT, value)

    def get_thin_pool_metadata_require_separate_pvs(self) -> str | None:
        """Get thin pool metadata require separate PVs setting.

        Returns:
            Setting value ('0' or '1'), or None if not found
        """
        return self.get(self.THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS)

    def set_thin_pool_metadata_require_separate_pvs(self, value: str) -> bool:
        """Set thin pool metadata require separate PVs setting.

        Args:
            value: Setting value ('0' or '1')

        Returns:
            True if successful, False otherwise
        """
        return self.set(self.THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS, value)

exists()

Check if the configuration file exists.

Returns:

Type	Description
bool	True if the configuration file exists, False otherwise

Source code in sts_libs/src/sts/lvm/lvconf.py

def exists(self) -> bool:
    """Check if the configuration file exists.

    Returns:
        True if the configuration file exists, False otherwise
    """
    return self.config_path.exists()

get(key)

Get configuration value from lvm.conf.

Searches for the specified key in lvm.conf and returns its value. Handles both commented (default) and uncommented (active) values. If both exist, returns the uncommented (active) value.

Parameters:

Name	Type	Description	Default
key	str	Configuration key to look up	required

Returns:

Type	Description
str | None	Configuration value if found, None otherwise

Example

config = LvmConfig()
threshold = config.get('thin_pool_autoextend_threshold')
# Returns '100' (default value, even if commented)

Source code in sts_libs/src/sts/lvm/lvconf.py

def get(self, key: str) -> str | None:
    """Get configuration value from lvm.conf.

    Searches for the specified key in lvm.conf and returns its value.
    Handles both commented (default) and uncommented (active) values.
    If both exist, returns the uncommented (active) value.

    Args:
        key: Configuration key to look up

    Returns:
        Configuration value if found, None otherwise

    Example:
        ```python
        config = LvmConfig()
        threshold = config.get('thin_pool_autoextend_threshold')
        # Returns '100' (default value, even if commented)
        ```
    """
    if not self.exists():
        logging.warning(f'LVM config file {self.config_path} not found')
        return None

    # Pattern to match both commented and uncommented key = value
    # Captures: (optional #) (whitespace) key (whitespace) = (whitespace) value
    # The value pattern handles the last occurrence of "key = value" format
    search_regex = re.compile(rf'^\s*#?\s*{re.escape(key)}\s*=\s*(\S+)')

    uncommented_value = None
    commented_value = None

    try:
        for line in self.config_path.read_text().splitlines():
            match = search_regex.match(line)
            if match:
                value = match.group(1)
                # Check if this is a commented line
                if line.strip().startswith('#'):
                    commented_value = value
                else:
                    uncommented_value = value
    except OSError:
        logging.exception(f'Failed to read LVM config file {self.config_path}')
        return None

    # Prefer uncommented (active) value over commented (default) value
    return uncommented_value if uncommented_value is not None else commented_value

get_thin_pool_autoextend_percent()

Get thin pool autoextend percent.

Returns:

Type	Description
str | None	Percent value as string, or None if not found

Source code in sts_libs/src/sts/lvm/lvconf.py

def get_thin_pool_autoextend_percent(self) -> str | None:
    """Get thin pool autoextend percent.

    Returns:
        Percent value as string, or None if not found
    """
    return self.get(self.THIN_POOL_AUTOEXTEND_PERCENT)

get_thin_pool_autoextend_threshold()

Get thin pool autoextend threshold.

Returns:

Type	Description
str | None	Threshold value as string, or None if not found

Source code in sts_libs/src/sts/lvm/lvconf.py

def get_thin_pool_autoextend_threshold(self) -> str | None:
    """Get thin pool autoextend threshold.

    Returns:
        Threshold value as string, or None if not found
    """
    return self.get(self.THIN_POOL_AUTOEXTEND_THRESHOLD)

get_thin_pool_metadata_require_separate_pvs()

Get thin pool metadata require separate PVs setting.

Returns:

Type	Description
str | None	Setting value ('0' or '1'), or None if not found

Source code in sts_libs/src/sts/lvm/lvconf.py

def get_thin_pool_metadata_require_separate_pvs(self) -> str | None:
    """Get thin pool metadata require separate PVs setting.

    Returns:
        Setting value ('0' or '1'), or None if not found
    """
    return self.get(self.THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS)

set(key, value)

Update a configuration value in lvm.conf.

Searches for the specified key in lvm.conf and updates its value. If the key is commented out, it will be uncommented. Preserves the original indentation.

Parameters:

Name	Type	Description	Default
key	str	Configuration key to update	required
value	str	New value to set	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

config = LvmConfig()
config.set('thin_pool_autoextend_threshold', '80')

Source code in sts_libs/src/sts/lvm/lvconf.py

def set(self, key: str, value: str) -> bool:
    """Update a configuration value in lvm.conf.

    Searches for the specified key in lvm.conf and updates its value.
    If the key is commented out, it will be uncommented.
    Preserves the original indentation.

    Args:
        key: Configuration key to update
        value: New value to set

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        config = LvmConfig()
        config.set('thin_pool_autoextend_threshold', '80')
        ```
    """
    if not self.exists():
        logging.error(f'LVM config file {self.config_path} not found')
        return False

    # Pattern to match both commented and uncommented key = value
    # Captures leading whitespace and optional comment marker
    search_regex = re.compile(rf'^(\s*)(#?\s*){re.escape(key)}(\s*)=(\s*)\S*')

    try:
        lines = self.config_path.read_text().splitlines()
        updated_lines = []
        found = False

        for line in lines:
            match = search_regex.match(line)
            if match and not found:
                # Get the leading whitespace (indentation)
                indent = match.group(1)
                # Create new line without comment marker
                updated_lines.append(f'{indent}{key} = {value}')
                found = True
            else:
                updated_lines.append(line)

        if not found:
            logging.warning(f'Configuration key {key} not found in {self.config_path}')
            return False

        self.config_path.write_text('\n'.join(updated_lines) + '\n')

    except OSError:
        logging.exception(f'Failed to update LVM config file {self.config_path}')
        return False

    return True

set_multiple(settings)

Update multiple configuration values in lvm.conf.

Updates all specified key-value pairs in a single operation. More efficient than calling set() multiple times. If keys are commented out, they will be uncommented.

Parameters:

Name	Type	Description	Default
settings	dict[str, str]	Dictionary of key-value pairs to update	required

Returns:

Type	Description
bool	True if all updates were successful, False otherwise

Example

config = LvmConfig()
config.set_multiple(
    {
        'thin_pool_autoextend_threshold': '80',
        'thin_pool_autoextend_percent': '50',
    }
)

Source code in sts_libs/src/sts/lvm/lvconf.py

def set_multiple(self, settings: dict[str, str]) -> bool:
    """Update multiple configuration values in lvm.conf.

    Updates all specified key-value pairs in a single operation.
    More efficient than calling set() multiple times.
    If keys are commented out, they will be uncommented.

    Args:
        settings: Dictionary of key-value pairs to update

    Returns:
        True if all updates were successful, False otherwise

    Example:
        ```python
        config = LvmConfig()
        config.set_multiple(
            {
                'thin_pool_autoextend_threshold': '80',
                'thin_pool_autoextend_percent': '50',
            }
        )
        ```
    """
    if not self.exists():
        logging.error(f'LVM config file {self.config_path} not found')
        return False

    try:
        lines = self.config_path.read_text().splitlines()
        updated_lines = []
        keys_found: set[str] = set()

        for line in lines:
            updated_line = line
            for key, value in settings.items():
                if key in keys_found:
                    continue
                # Pattern to match both commented and uncommented key = value
                search_regex = re.compile(rf'^(\s*)(#?\s*){re.escape(key)}(\s*)=(\s*)\S*')
                match = search_regex.match(line)
                if match:
                    # Get the leading whitespace (indentation)
                    indent = match.group(1)
                    # Create new line without comment marker
                    updated_line = f'{indent}{key} = {value}'
                    keys_found.add(key)
                    break
            updated_lines.append(updated_line)

        # Check if all keys were found
        missing_keys = set(settings.keys()) - keys_found
        if missing_keys:
            logging.warning(f'Configuration keys not found: {missing_keys}')
            return False

        self.config_path.write_text('\n'.join(updated_lines) + '\n')

    except OSError:
        logging.exception(f'Failed to update LVM config file {self.config_path}')
        return False

    return True

set_thin_pool_autoextend_percent(value)

Set thin pool autoextend percent.

Parameters:

Name	Type	Description	Default
value	str	Percent value	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/lvm/lvconf.py

def set_thin_pool_autoextend_percent(self, value: str) -> bool:
    """Set thin pool autoextend percent.

    Args:
        value: Percent value

    Returns:
        True if successful, False otherwise
    """
    return self.set(self.THIN_POOL_AUTOEXTEND_PERCENT, value)

set_thin_pool_autoextend_threshold(value)

Set thin pool autoextend threshold.

Parameters:

Name	Type	Description	Default
value	str	Threshold value (0-100)	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/lvm/lvconf.py

def set_thin_pool_autoextend_threshold(self, value: str) -> bool:
    """Set thin pool autoextend threshold.

    Args:
        value: Threshold value (0-100)

    Returns:
        True if successful, False otherwise
    """
    return self.set(self.THIN_POOL_AUTOEXTEND_THRESHOLD, value)

set_thin_pool_metadata_require_separate_pvs(value)

Set thin pool metadata require separate PVs setting.

Parameters:

Name	Type	Description	Default
value	str	Setting value ('0' or '1')	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/lvm/lvconf.py

def set_thin_pool_metadata_require_separate_pvs(self, value: str) -> bool:
    """Set thin pool metadata require separate PVs setting.

    Args:
        value: Setting value ('0' or '1')

    Returns:
        True if successful, False otherwise
    """
    return self.set(self.THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS, value)

LvmDevice dataclass

Bases: StorageDevice

Base class for LVM devices.

Provides common functionality for all LVM device types: - Command execution with standard options - Configuration management - Basic device operations

Parameters:

Name	Type	Description	Default
name	str | None	Device name (optional)	None
path	Path | str | None	Device path (optional, defaults to /dev/)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
model	str | None	Device model (optional)	None
yes	bool	Automatically answer yes to prompts	True
force	bool	Force operations without confirmation	False

The yes and force options are useful for automation: - yes: Skip interactive prompts - force: Ignore warnings and errors

Source code in sts_libs/src/sts/lvm/base.py

@dataclass
class LvmDevice(StorageDevice):
    """Base class for LVM devices.

    Provides common functionality for all LVM device types:
    - Command execution with standard options
    - Configuration management
    - Basic device operations

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation

    The yes and force options are useful for automation:
    - yes: Skip interactive prompts
    - force: Ignore warnings and errors
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: Path | str | None = None
    size: int | None = None
    model: str | None = None
    validate_on_init = False

    # Optional parameters for this class
    yes: bool = True  # Answer yes to prompts
    force: bool = False  # Force operations

    # Internal fields
    config_path: Path | str = Path('/etc/lvm/lvm.conf')

    def __post_init__(self) -> None:
        """Initialize LVM device."""
        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/{self.name}'

        # Initialize parent class
        super().__post_init__()

    def _run(self, cmd: str, *args: str | Path | None, **kwargs: str) -> CommandResult:
        """Run LVM command.

        Builds and executes LVM commands with standard options:
        - Adds --yes for non-interactive mode
        - Adds --force to ignore warnings
        - Converts Python parameters to LVM options

        Args:
            cmd: Command name (e.g. 'pvcreate')
            *args: Command arguments
            **kwargs: Command parameters

        Returns:
            Command result
        """
        command = [cmd]
        if self.yes:
            command.append('--yes')
        if self.force:
            command.append('--force')
        if args:
            command.extend(str(arg) for arg in args if arg)
        if kwargs:
            command.extend(f'--{k.replace("_", "-")}={v}' for k, v in kwargs.items() if v)

        return run(' '.join(command))

    @abstractmethod
    def create(self, **options: str) -> bool:
        """Create LVM device.

        Args:
            **options: Device options (see LvmOptions)

        Returns:
            True if successful, False otherwise
        """

    @abstractmethod
    def remove(self, **options: str) -> bool:
        """Remove LVM device.

        Args:
            **options: Device options (see LvmOptions)

        Returns:
            True if successful, False otherwise
        """

__post_init__()

Initialize LVM device.

Source code in sts_libs/src/sts/lvm/base.py

def __post_init__(self) -> None:
    """Initialize LVM device."""
    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/{self.name}'

    # Initialize parent class
    super().__post_init__()

create(**options) abstractmethod

Create LVM device.

Parameters:

Name	Type	Description	Default
**options	str	Device options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/lvm/base.py

@abstractmethod
def create(self, **options: str) -> bool:
    """Create LVM device.

    Args:
        **options: Device options (see LvmOptions)

    Returns:
        True if successful, False otherwise
    """

remove(**options) abstractmethod

Remove LVM device.

Parameters:

Name	Type	Description	Default
**options	str	Device options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/lvm/base.py

@abstractmethod
def remove(self, **options: str) -> bool:
    """Remove LVM device.

    Args:
        **options: Device options (see LvmOptions)

    Returns:
        True if successful, False otherwise
    """

PhysicalVolume dataclass

Bases: LvmDevice

Physical Volume device.

A Physical Volume (PV) is a disk or partition used by LVM. PVs provide the storage pool for Volume Groups.

Key features: - Initialize disks/partitions for LVM use - Track space allocation - Handle bad block management - Store LVM metadata

Parameters:

Name	Type	Description	Default
name	str | None	Device name (optional)	None
path	Path | str | None	Device path (optional, defaults to /dev/)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
model	str | None	Device model (optional)	None
yes	bool	Automatically answer yes to prompts	True
force	bool	Force operations without confirmation	False
vg	str | None	Volume group name (optional, discovered from device)	None
fmt	str | None	PV format (optional, discovered from device)	None
attr	str | None	PV attributes (optional, discovered from device)	None
pfree	str | None	PV free space (optional, discovered from device)	None

Example

pv = PhysicalVolume(name='sda1')  # Discovers other values
pv = PhysicalVolume.create('/dev/sda1')  # Creates new PV

Source code in sts_libs/src/sts/lvm/physical_volume.py

@dataclass
class PhysicalVolume(LvmDevice):
    """Physical Volume device.

    A Physical Volume (PV) is a disk or partition used by LVM.
    PVs provide the storage pool for Volume Groups.

    Key features:
    - Initialize disks/partitions for LVM use
    - Track space allocation
    - Handle bad block management
    - Store LVM metadata

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        vg: Volume group name (optional, discovered from device)
        fmt: PV format (optional, discovered from device)
        attr: PV attributes (optional, discovered from device)
        pfree: PV free space (optional, discovered from device)

    Example:
        ```python
        pv = PhysicalVolume(name='sda1')  # Discovers other values
        pv = PhysicalVolume.create('/dev/sda1')  # Creates new PV
        ```
    """

    # Optional parameters for this class
    vg: str | None = None  # Volume Group membership
    fmt: str | None = None  # PV format (usually lvm2)
    attr: str | None = None  # PV attributes
    pfree: str | None = None  # Free space

    # Available PV commands
    COMMANDS: ClassVar[list[str]] = [
        'pvchange',  # Modify PV attributes
        'pvck',  # Check PV metadata
        'pvcreate',  # Initialize PV
        'pvdisplay',  # Show PV details
        'pvmove',  # Move PV data
        'pvremove',  # Remove PV
        'pvresize',  # Resize PV
        'pvs',  # List PVs
        'pvscan',  # Scan for PVs
    ]

    # Discover PV info if path is available
    def discover_pv_info(self) -> None:
        """Discovers PV information if path is available.

        Volume group membership.
        Format and attributes.
        Size information.
        """
        result = run(f'pvs {self.path} --noheadings --separator ","')
        if result.succeeded:
            # Parse PV info line
            # Format: PV,VG,Fmt,Attr,PSize,PFree
            parts = result.stdout.strip().split(',')
            if len(parts) == 6:
                _, vg, fmt, attr, _, pfree = parts
                if not self.vg:
                    self.vg = vg or None
                if not self.fmt:
                    self.fmt = fmt
                if not self.attr:
                    self.attr = attr
                if not self.pfree:
                    self.pfree = pfree

    def create(self, **options: str) -> bool:
        """Create Physical Volume.

        Initializes a disk or partition for use with LVM:
        - Creates LVM metadata area
        - Prepares device for VG membership

        Args:
            **options: PV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            pv = PhysicalVolume(path='/dev/sda1')
            pv.create()
            True
            ```
        """
        if not self.path:
            logging.error('Device path required')
            return False

        result = self._run('pvcreate', str(self.path), **options)
        if result.succeeded:
            self.discover_pv_info()
        return result.succeeded

    def remove(self, **options: str) -> bool:
        """Remove Physical Volume.

        Removes LVM metadata from device:
        - Device must not be in use by a VG
        - Data on device is not erased

        Args:
            **options: PV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            pv = PhysicalVolume(path='/dev/sda1')
            pv.remove()
            True
            ```
        """
        if not self.path:
            logging.error('Device path required')
            return False

        result = self._run('pvremove', str(self.path), **options)
        return result.succeeded

    @classmethod
    def from_blockdevice(cls, block_device: BlockDevice, *, create: bool = False, **options: str) -> PhysicalVolume:
        """Create PhysicalVolume from BlockDevice.

        Creates a PhysicalVolume instance from an existing BlockDevice,
        optionally initializing it with pvcreate.

        Args:
            block_device: BlockDevice instance to create PV from
            create: Whether to run pvcreate to initialize the device
            **options: Additional pvcreate options (see pvcreate man page)

        Returns:
            PhysicalVolume instance

        Raises:
            ValueError: If block_device is None or has no path
            RuntimeError: If create=True and pvcreate fails

        Example:
            ```python
            from sts.blockdevice import BlockDevice

            # Create PV without initializing
            bd = BlockDevice('/dev/sda1')
            pv = PhysicalVolume.from_blockdevice(bd)

            # Create and initialize PV
            pv = PhysicalVolume.from_blockdevice(bd, create=True)

            # Create with options
            pv = PhysicalVolume.from_blockdevice(bd, create=True, dataalignment='1m')
            ```
        """
        if not block_device:
            msg = 'BlockDevice instance is required'
            raise ValueError(msg)

        if not block_device.path:
            msg = 'BlockDevice must have a valid path'
            raise ValueError(msg)

        # Create PhysicalVolume instance with attributes from BlockDevice
        pv = cls(
            path=str(block_device.path),
            size=block_device.size,
            model=block_device.model,
        )

        # Initialize the device if requested
        if create:
            success = pv.create(**options)
            if not success:
                msg = f'Failed to create physical volume on {block_device.path}'
                raise RuntimeError(msg)

        return pv

    @classmethod
    def get_all(cls) -> dict[str, PVInfo]:
        """Get all Physical Volumes.

        Returns:
            Dictionary mapping PV names to their information

        Example:
            ```python
            PhysicalVolume.get_all()
            {'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}
            ```
        """
        result = run('pvs --noheadings --separator ","')
        if result.failed:
            logging.debug('No Physical Volumes found')
            return {}

        # Format: PV,VG,Fmt,Attr,PSize,PFree
        pv_info_regex = r'\s+(\S+),(\S+)?,(\S+),(.*),(.*),(.*)$'
        pv_dict = {}

        for line in result.stdout.splitlines():
            if match := re.match(pv_info_regex, line):
                pv_dict[match.group(1)] = PVInfo(
                    vg=match.group(2) or None,  # VG can be empty
                    fmt=match.group(3),
                    attr=match.group(4),
                    psize=match.group(5),
                    pfree=match.group(6),
                )

        return pv_dict

create(**options)

Create Physical Volume.

Initializes a disk or partition for use with LVM: - Creates LVM metadata area - Prepares device for VG membership

Parameters:

Name	Type	Description	Default
**options	str	PV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

pv = PhysicalVolume(path='/dev/sda1')
pv.create()
True

Source code in sts_libs/src/sts/lvm/physical_volume.py

def create(self, **options: str) -> bool:
    """Create Physical Volume.

    Initializes a disk or partition for use with LVM:
    - Creates LVM metadata area
    - Prepares device for VG membership

    Args:
        **options: PV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        pv = PhysicalVolume(path='/dev/sda1')
        pv.create()
        True
        ```
    """
    if not self.path:
        logging.error('Device path required')
        return False

    result = self._run('pvcreate', str(self.path), **options)
    if result.succeeded:
        self.discover_pv_info()
    return result.succeeded

discover_pv_info()

Discovers PV information if path is available.

Volume group membership. Format and attributes. Size information.

Source code in sts_libs/src/sts/lvm/physical_volume.py

def discover_pv_info(self) -> None:
    """Discovers PV information if path is available.

    Volume group membership.
    Format and attributes.
    Size information.
    """
    result = run(f'pvs {self.path} --noheadings --separator ","')
    if result.succeeded:
        # Parse PV info line
        # Format: PV,VG,Fmt,Attr,PSize,PFree
        parts = result.stdout.strip().split(',')
        if len(parts) == 6:
            _, vg, fmt, attr, _, pfree = parts
            if not self.vg:
                self.vg = vg or None
            if not self.fmt:
                self.fmt = fmt
            if not self.attr:
                self.attr = attr
            if not self.pfree:
                self.pfree = pfree

from_blockdevice(block_device, *, create=False, **options) classmethod

Create PhysicalVolume from BlockDevice.

Creates a PhysicalVolume instance from an existing BlockDevice, optionally initializing it with pvcreate.

Parameters:

Name	Type	Description	Default
block_device	BlockDevice	BlockDevice instance to create PV from	required
create	bool	Whether to run pvcreate to initialize the device	False
**options	str	Additional pvcreate options (see pvcreate man page)	{}

Returns:

Type	Description
PhysicalVolume	PhysicalVolume instance

Raises:

Type	Description
ValueError	If block_device is None or has no path
RuntimeError	If create=True and pvcreate fails

Example

from sts.blockdevice import BlockDevice

# Create PV without initializing
bd = BlockDevice('/dev/sda1')
pv = PhysicalVolume.from_blockdevice(bd)

# Create and initialize PV
pv = PhysicalVolume.from_blockdevice(bd, create=True)

# Create with options
pv = PhysicalVolume.from_blockdevice(bd, create=True, dataalignment='1m')

Source code in sts_libs/src/sts/lvm/physical_volume.py

@classmethod
def from_blockdevice(cls, block_device: BlockDevice, *, create: bool = False, **options: str) -> PhysicalVolume:
    """Create PhysicalVolume from BlockDevice.

    Creates a PhysicalVolume instance from an existing BlockDevice,
    optionally initializing it with pvcreate.

    Args:
        block_device: BlockDevice instance to create PV from
        create: Whether to run pvcreate to initialize the device
        **options: Additional pvcreate options (see pvcreate man page)

    Returns:
        PhysicalVolume instance

    Raises:
        ValueError: If block_device is None or has no path
        RuntimeError: If create=True and pvcreate fails

    Example:
        ```python
        from sts.blockdevice import BlockDevice

        # Create PV without initializing
        bd = BlockDevice('/dev/sda1')
        pv = PhysicalVolume.from_blockdevice(bd)

        # Create and initialize PV
        pv = PhysicalVolume.from_blockdevice(bd, create=True)

        # Create with options
        pv = PhysicalVolume.from_blockdevice(bd, create=True, dataalignment='1m')
        ```
    """
    if not block_device:
        msg = 'BlockDevice instance is required'
        raise ValueError(msg)

    if not block_device.path:
        msg = 'BlockDevice must have a valid path'
        raise ValueError(msg)

    # Create PhysicalVolume instance with attributes from BlockDevice
    pv = cls(
        path=str(block_device.path),
        size=block_device.size,
        model=block_device.model,
    )

    # Initialize the device if requested
    if create:
        success = pv.create(**options)
        if not success:
            msg = f'Failed to create physical volume on {block_device.path}'
            raise RuntimeError(msg)

    return pv

get_all() classmethod

Get all Physical Volumes.

Returns:

Type	Description
dict[str, PVInfo]	Dictionary mapping PV names to their information

Example

PhysicalVolume.get_all()
{'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}

Source code in sts_libs/src/sts/lvm/physical_volume.py

@classmethod
def get_all(cls) -> dict[str, PVInfo]:
    """Get all Physical Volumes.

    Returns:
        Dictionary mapping PV names to their information

    Example:
        ```python
        PhysicalVolume.get_all()
        {'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}
        ```
    """
    result = run('pvs --noheadings --separator ","')
    if result.failed:
        logging.debug('No Physical Volumes found')
        return {}

    # Format: PV,VG,Fmt,Attr,PSize,PFree
    pv_info_regex = r'\s+(\S+),(\S+)?,(\S+),(.*),(.*),(.*)$'
    pv_dict = {}

    for line in result.stdout.splitlines():
        if match := re.match(pv_info_regex, line):
            pv_dict[match.group(1)] = PVInfo(
                vg=match.group(2) or None,  # VG can be empty
                fmt=match.group(3),
                attr=match.group(4),
                psize=match.group(5),
                pfree=match.group(6),
            )

    return pv_dict

remove(**options)

Remove Physical Volume.

Removes LVM metadata from device: - Device must not be in use by a VG - Data on device is not erased

Parameters:

Name	Type	Description	Default
**options	str	PV options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

pv = PhysicalVolume(path='/dev/sda1')
pv.remove()
True

Source code in sts_libs/src/sts/lvm/physical_volume.py

def remove(self, **options: str) -> bool:
    """Remove Physical Volume.

    Removes LVM metadata from device:
    - Device must not be in use by a VG
    - Data on device is not erased

    Args:
        **options: PV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        pv = PhysicalVolume(path='/dev/sda1')
        pv.remove()
        True
        ```
    """
    if not self.path:
        logging.error('Device path required')
        return False

    result = self._run('pvremove', str(self.path), **options)
    return result.succeeded

ThinPool dataclass

Bases: LogicalVolume

Thin Pool logical volume.

A Thin Pool is a special type of logical volume that provides thin provisioning capabilities. It manages multiple thin volumes and consists of two special components: - tdata: The data component that stores actual data - tmeta: The metadata component that stores thin provisioning metadata

The thin pool can hold multiple logical volumes (thin volumes) that share the pool's space dynamically.

Parameters:

Name	Type	Description	Default
name	str | None	Pool name (required)	None
vg	str | None	Volume group name (required)	None
path	Path | str | None	Device path (optional, defaults to /dev//)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
model	str | None	Device model (optional)	None
yes	bool	Automatically answer yes to prompts	True
force	bool	Force operations without confirmation	False
report	LVReport | None	LV report instance (optional, created automatically)	None
prevent_report_updates	bool	Prevent automatic report updates (defaults to False)	False
thin_volumes	list[LogicalVolume]	List of thin volumes in this pool (discovered automatically)	list()

Key features: - Thin provisioning with overcommitment - Automatic space allocation - Snapshot support - Pool usage monitoring - Component conversion (tdata/tmeta)

Example

# Create thin pool
pool = ThinPool(name='pool1', vg='vg0')
pool.create(size='1G')

# Create thin volumes in the pool
thin_lv1 = pool.create_thin_volume('thin1', virtualsize='500M')
thin_lv2 = pool.create_thin_volume('thin2', virtualsize='800M')

# Monitor pool usage
data_usage, meta_usage = pool.get_pool_usage()

# Access special volumes
tdata = pool.get_tdata_volume()
tmeta = pool.get_tmeta_volume()

Source code in sts_libs/src/sts/lvm/logical_volume.py

@dataclass
class ThinPool(LogicalVolume):
    """Thin Pool logical volume.

    A Thin Pool is a special type of logical volume that provides thin provisioning
    capabilities. It manages multiple thin volumes and consists of two special
    components:
    - tdata: The data component that stores actual data
    - tmeta: The metadata component that stores thin provisioning metadata

    The thin pool can hold multiple logical volumes (thin volumes) that share
    the pool's space dynamically.

    Args:
        name: Pool name (required)
        vg: Volume group name (required)
        path: Device path (optional, defaults to /dev/<vg>/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        report: LV report instance (optional, created automatically)
        prevent_report_updates: Prevent automatic report updates (defaults to False)
        thin_volumes: List of thin volumes in this pool (discovered automatically)

    Key features:
    - Thin provisioning with overcommitment
    - Automatic space allocation
    - Snapshot support
    - Pool usage monitoring
    - Component conversion (tdata/tmeta)

    Example:
        ```python
        # Create thin pool
        pool = ThinPool(name='pool1', vg='vg0')
        pool.create(size='1G')

        # Create thin volumes in the pool
        thin_lv1 = pool.create_thin_volume('thin1', virtualsize='500M')
        thin_lv2 = pool.create_thin_volume('thin2', virtualsize='800M')

        # Monitor pool usage
        data_usage, meta_usage = pool.get_pool_usage()

        # Access special volumes
        tdata = pool.get_tdata_volume()
        tmeta = pool.get_tmeta_volume()
        ```
    """

    # Thin volumes in this pool
    thin_volumes: list[LogicalVolume] = field(default_factory=list, repr=False)
    tdata: LogicalVolume | None = None
    tmeta: LogicalVolume | None = None

    def __post_init__(self) -> None:
        """Initialize Thin Pool.

        - Sets device path from name and VG
        - Initializes parent LogicalVolume
        - Discovers thin volumes in the pool
        - Populates tdata and tmeta component references
        """
        super().__post_init__()

    # @override
    def refresh_report(self) -> bool:
        """Refresh Thin pool LV report data.

        Creates or updates the Thin pool LV report with the latest information.
        Also refreshes the tdata and tmeta component reports if they exist.

        Returns:
            bool: True if refresh was successful
        """
        # Create new report if needed
        if not self.report:
            # Do not provide name and vg during init to prevent update
            self.report = LVReport()
            self.report.name = self.name
            self.report.vg = self.vg

        # Refresh the pool report
        success = self.report.refresh()

        # Refresh tdata and tmeta if they exist
        if self.get_tdata_volume() and self.tdata:
            self.tdata.refresh_report()
        if self.get_tmeta_volume() and self.tmeta:
            self.tmeta.refresh_report()

        return success

    # @override
    def create(self, *args: str, **options: str) -> bool:
        """Create Thin Pool logical volume.

        Overrides the parent LogicalVolume.create() method to automatically
        set type='thin-pool' since ThinPool objects are always thin pools.

        Args:
            *args: Additional arguments passed to lvcreate
            **options: All options passed as strings to lvcreate command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            pool.create(size='1G')  # type='thin-pool' is automatically set

            # Can also specify additional options
            pool.create(size='1G', chunksize='256k', poolmetadatasize='4M')
            ```
        """
        # Always set type to 'thin-pool' for ThinPool objects
        # This ensures the correct LV type is created regardless of what user passes
        options['type'] = 'thin-pool'

        # Call parent create method
        return super().create(*args, **options)

    def discover_thin_volumes(self) -> list[LogicalVolume]:
        """Discover thin volumes in this pool.

        Returns:
            List of LogicalVolume instances representing thin volumes in this pool
        """
        self.thin_volumes = []

        # Get all LVs in the VG
        all_lvs = LogicalVolume.get_all(self.vg)

        # Filter for thin volumes that belong to this pool
        for lv in all_lvs:
            if (
                lv.report
                and lv.report.pool_lv
                and lv.report.pool_lv.strip('[]') == self.name
                and lv.report.lv_layout
                and 'thin' in lv.report.lv_layout
            ):
                self.thin_volumes.append(lv)

        return self.thin_volumes

    def get_tdata_volume(self) -> LogicalVolume | None:
        """Get the tdata (data) component of the thin pool.

        The tdata volume stores the actual data for all thin volumes in the pool.
        It's always named <pool_name>_tdata by LVM convention.

        Returns:
            LogicalVolume instance for the tdata component, or None if not found

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            tdata = pool.get_tdata_volume()
            if tdata:
                print(f'Data volume: {tdata.name}')
            ```
        """
        if not self.name or not self.vg:
            return None

        # LVM always names the data component as {pool_name}_tdata
        data_lv_name = f'{self.name}_tdata'

        if self.tdata:
            return self.tdata

        try:
            self.tdata = LogicalVolume(name=data_lv_name, vg=self.vg)
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to get tdata volume {data_lv_name}: {e}')
            return None
        else:
            return self.tdata

    def get_tmeta_volume(self) -> LogicalVolume | None:
        """Get the tmeta (metadata) component of the thin pool.

        The tmeta volume stores the thin provisioning metadata for the pool.
        It's always named <pool_name>_tmeta by LVM convention.

        Returns:
            LogicalVolume instance for the tmeta component, or None if not found

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            tmeta = pool.get_tmeta_volume()
            if tmeta:
                print(f'Metadata volume: {tmeta.name}')
            ```
        """
        if not self.name or not self.vg:
            return None

        # LVM always names the metadata component as {pool_name}_tmeta
        meta_lv_name = f'{self.name}_tmeta'

        if self.tmeta:
            return self.tmeta

        try:
            self.tmeta = LogicalVolume(name=meta_lv_name, vg=self.vg)
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to get tmeta volume {meta_lv_name}: {e}')
            return None
        else:
            return self.tmeta

    def convert_pool_data(self, *args: str, **options: str) -> bool:
        """Convert thin pool data component to specified type.

        Converts the thin pool's data component (e.g., from linear to RAID1).
        This is typically used to add mirroring or change the RAID level of the pool's data.

        Args:
            *args: Additional arguments passed to lvconvert
            **options: All options passed as strings to lvconvert command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Convert pool data to RAID1 with 3 mirrors
            pool.convert_pool_data(type='raid1', mirrors='3')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        # Convert the pool's data component (always named {pool_name}_tdata)
        result = self._run('lvconvert', f'{self.vg}/{self.name}_tdata', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_pool_metadata(self, *args: str, **options: str) -> bool:
        """Convert thin pool metadata component to specified type.

        Converts the thin pool's metadata component (e.g., from linear to RAID1).
        This is typically used to add mirroring or change the RAID level of the pool's metadata.

        Args:
            *args: Additional arguments passed to lvconvert
            **options: All options passed as strings to lvconvert command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Convert pool metadata to RAID1 with 1 mirror
            pool.convert_pool_metadata(type='raid1', mirrors='1')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        # Convert the pool's metadata component (always named {pool_name}_tmeta)
        result = self._run('lvconvert', f'{self.vg}/{self.name}_tmeta', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def swap_metadata(self, metadata_lv: str, *args: str, **options: str) -> bool:
        """Swap thin pool metadata with another LV.

        Swaps the thin pool's metadata device with the specified LV.
        This is useful for metadata repair operations where you need to
        extract, check, or replace the metadata device.

        The command executed is:
            lvconvert --thinpool <vg>/<pool> --poolmetadata <metadata_lv>

        Args:
            metadata_lv: The LV to swap with the pool's metadata (format: vg/lv or just lv)
            *args: Additional arguments passed to lvconvert
            **options: All options passed as strings to lvconvert command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Create a temporary LV for metadata swap
            meta_swap = LogicalVolume(name='meta_swap', vg='vg0')
            meta_swap.create(size='4M')

            # Swap metadata - the pool's current metadata goes to meta_swap
            pool.swap_metadata('vg0/meta_swap')

            # Now meta_swap contains the old metadata, pool has new empty metadata
            # You can activate meta_swap and run thin_check on it

            # Swap back
            pool.swap_metadata('vg0/meta_swap')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not metadata_lv:
            logging.error('Metadata LV path required')
            return False

        # Build the command with --thinpool option
        result = self._run(
            'lvconvert',
            '--thinpool',
            f'{self.vg}/{self.name}',
            '--poolmetadata',
            metadata_lv,
            *args,
            **options,
        )
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def repair(self, pv_device: str | None = None, *args: str, **options: str) -> bool:
        """Repair thin pool metadata.

        Attempts to repair the thin pool's metadata using lvconvert --repair.
        This creates a new metadata device and attempts to recover the pool.

        Args:
            pv_device: Optional PV device to use for the new metadata
            *args: Additional arguments passed to lvconvert
            **options: All options passed as strings to lvconvert command

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Repair pool using any available space
            pool.repair()

            # Repair pool using specific PV for new metadata
            pool.repair('/dev/sdb')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        cmd_args = ['--repair', f'{self.vg}/{self.name}']
        if pv_device:
            cmd_args.append(pv_device)
        if args:
            cmd_args.extend(args)

        result = self._run('lvconvert', *cmd_args, **options)
        logging.info(result.stdout)
        logging.info(result.stderr)
        logging.info(result.rc)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def get_pool_usage(self) -> tuple[str, str]:
        """Get thin pool data and metadata usage percentages.

        Returns:
            Tuple of (data_percent, metadata_percent) as strings

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            data_usage, meta_usage = pool.get_pool_usage()
            print(f'Data usage: {data_usage}%, Metadata usage: {meta_usage}%')
            ```
        """
        if not self.vg or not self.name:
            logging.error('Volume group and logical volume name required')
            return 'unknown', 'unknown'

        # Get data usage
        data_result = self.lvs(f'{self.vg}/{self.name}', o='data_percent', noheadings='')
        data_percent = data_result.stdout.strip() if data_result.succeeded else 'unknown'

        # Get metadata usage
        meta_result = self.lvs(f'{self.vg}/{self.name}', o='metadata_percent', noheadings='')
        meta_percent = meta_result.stdout.strip() if meta_result.succeeded else 'unknown'

        return data_percent, meta_percent

    def get_data_stripes(self) -> str | None:
        """Get stripe count for thin pool data component.

        For thin pools, the actual stripe information is stored in the data component (_tdata),
        not in the main pool LV. This method returns the stripe count from the data component.

        Returns:
            String representing stripe count, or None if not a thin pool or stripes not found

        Example:
            ```python
            pool = ThinPool(name='pool', vg='vg0')
            pool.create(stripes='2', size='100M')
            stripe_count = pool.get_data_stripes()  # Returns '2'
            ```
        """
        assert self.refresh_report()
        if self.tdata and self.tdata.report:
            return self.tdata.report.stripes
        return None

    def get_data_stripe_size(self) -> str | None:
        """Get stripe size for thin pool data component.

        For thin pools, the actual stripe size information is stored in the data component (_tdata),
        not in the main pool LV. This method returns the stripe size from the data component.

        Returns:
            String representing stripe size, or None if not a thin pool or stripe size not found

        Example:
            ```python
            pool = ThinPool(name='pool', vg='vg0')
            pool.create(stripes='2', stripesize='64k', size='100M')
            stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'
            ```
        """
        assert self.refresh_report()
        if self.tdata and self.tdata.report:
            return self.tdata.report.stripe_size
        return None

    def create_thin_volume(self, lv_name: str, *args: str, **options: str) -> LogicalVolume:
        """Create thin volume in this pool.

        Args:
            lv_name: Thin volume name
            *args: Additional arguments passed to lvcreate
            **options: All options passed as strings to lvcreate command

        Returns:
            LogicalVolume object for the created thin volume

        Raises:
            AssertionError: If volume creation fails

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            thin_lv = pool.create_thin_volume('thin1', virtualsize='500M')
            ```
        """
        thin_lv = LogicalVolume(name=lv_name, pool_name=self.name, vg=self.vg)
        assert thin_lv.create(*args, **options), f'Failed to create thin volume {lv_name}'
        assert self.refresh_report()

        # Add to our thin volumes list
        self.thin_volumes.append(thin_lv)

        return thin_lv

    def get_thin_volume_count(self) -> int:
        """Get the number of thin volumes in this pool.

        Returns:
            Number of thin volumes in the pool
        """
        if self.report and self.report.thin_count:
            try:
                return int(self.report.thin_count)
            except ValueError:
                pass

        # Fallback to counting discovered volumes
        return len(self.thin_volumes)

    @classmethod
    def create_thin_pool(cls, pool_name: str, vg_name: str, *args: str, **options: str) -> ThinPool:
        """Create thin pool with specified options.

        Args:
            pool_name: Pool name
            vg_name: Volume group name
            *args: Additional arguments passed to lvcreate
            **options: All options passed as strings to lvcreate command

        Returns:
            ThinPool object for the created pool

        Raises:
            AssertionError: If pool creation fails

        Example:
            ```python
            pool = ThinPool.create_thin_pool('pool1', 'vg0', size='1G')
            ```
        """
        pool = cls(name=pool_name, vg=vg_name)
        assert pool.create(*args, **options), f'Failed to create thin pool {pool_name}'
        return pool

    def remove_thin_volumes(self, *, force: bool = True) -> bool:
        """Remove all thin volumes in this pool.

        This method discovers and removes all thin volumes that belong to this pool.
        It should be called before removing the pool itself.

        Args:
            force: If True, use force flags for removal (default: True)

        Returns:
            True if all thin volumes were removed successfully, False otherwise

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            pool.remove_thin_volumes()  # Remove all thin volumes first
            pool.remove()  # Now remove the pool
            ```
        """
        # Discover thin volumes in this pool
        self.discover_thin_volumes()

        if not self.thin_volumes:
            logging.info(f'No thin volumes found in pool {self.name}')
            return True

        all_removed = True
        for thin_lv in self.thin_volumes:
            logging.info(f'Removing thin volume {thin_lv.name} from pool {self.name}')
            success = thin_lv.remove(force='', yes='') if force else thin_lv.remove()

            if not success:
                logging.warning(f'Failed to remove thin volume {thin_lv.name}')
                all_removed = False

        # Clear the thin volumes list after removal
        if all_removed:
            self.thin_volumes = []

        return all_removed

    def remove_with_thin_volumes(self, *args: str, **options: str) -> bool:
        """Remove thin pool along with all its thin volumes.

        This is a convenience method that handles the proper cleanup order:
        1. Remove all thin volumes in the pool
        2. Remove the pool itself

        Args:
            *args: Additional arguments passed to lvremove
            **options: All options passed as strings to lvremove command

        Returns:
            True if pool and all thin volumes were removed successfully

        Example:
            ```python
            pool = ThinPool(name='pool1', vg='vg0')
            pool.create(size='1G')
            pool.create_thin_volume('thin1', virtualsize='500M')
            pool.remove_with_thin_volumes(force='', yes='')  # Removes everything
            ```
        """
        # First remove all thin volumes
        if not self.remove_thin_volumes(force='force' in options or 'yes' in options):
            logging.error(f'Failed to remove thin volumes from pool {self.name}')
            return False

        # Then remove the pool
        return self.remove(*args, **options)

    def __str__(self) -> str:
        """String representation of ThinPool."""
        thin_count = self.get_thin_volume_count()
        return f"ThinPool(name='{self.name}', vg='{self.vg}', thin_volumes={thin_count})"

__post_init__()

Initialize Thin Pool.

• Sets device path from name and VG
• Initializes parent LogicalVolume
• Discovers thin volumes in the pool
• Populates tdata and tmeta component references

Source code in sts_libs/src/sts/lvm/logical_volume.py

def __post_init__(self) -> None:
    """Initialize Thin Pool.

    - Sets device path from name and VG
    - Initializes parent LogicalVolume
    - Discovers thin volumes in the pool
    - Populates tdata and tmeta component references
    """
    super().__post_init__()

__str__()

String representation of ThinPool.

Source code in sts_libs/src/sts/lvm/logical_volume.py

def __str__(self) -> str:
    """String representation of ThinPool."""
    thin_count = self.get_thin_volume_count()
    return f"ThinPool(name='{self.name}', vg='{self.vg}', thin_volumes={thin_count})"

convert_pool_data(*args, **options)

Convert thin pool data component to specified type.

Converts the thin pool's data component (e.g., from linear to RAID1). This is typically used to add mirroring or change the RAID level of the pool's data.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvconvert	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Convert pool data to RAID1 with 3 mirrors
pool.convert_pool_data(type='raid1', mirrors='3')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert_pool_data(self, *args: str, **options: str) -> bool:
    """Convert thin pool data component to specified type.

    Converts the thin pool's data component (e.g., from linear to RAID1).
    This is typically used to add mirroring or change the RAID level of the pool's data.

    Args:
        *args: Additional arguments passed to lvconvert
        **options: All options passed as strings to lvconvert command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Convert pool data to RAID1 with 3 mirrors
        pool.convert_pool_data(type='raid1', mirrors='3')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    # Convert the pool's data component (always named {pool_name}_tdata)
    result = self._run('lvconvert', f'{self.vg}/{self.name}_tdata', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

convert_pool_metadata(*args, **options)

Convert thin pool metadata component to specified type.

Converts the thin pool's metadata component (e.g., from linear to RAID1). This is typically used to add mirroring or change the RAID level of the pool's metadata.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvconvert	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Convert pool metadata to RAID1 with 1 mirror
pool.convert_pool_metadata(type='raid1', mirrors='1')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def convert_pool_metadata(self, *args: str, **options: str) -> bool:
    """Convert thin pool metadata component to specified type.

    Converts the thin pool's metadata component (e.g., from linear to RAID1).
    This is typically used to add mirroring or change the RAID level of the pool's metadata.

    Args:
        *args: Additional arguments passed to lvconvert
        **options: All options passed as strings to lvconvert command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Convert pool metadata to RAID1 with 1 mirror
        pool.convert_pool_metadata(type='raid1', mirrors='1')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    # Convert the pool's metadata component (always named {pool_name}_tmeta)
    result = self._run('lvconvert', f'{self.vg}/{self.name}_tmeta', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

create(*args, **options)

Create Thin Pool logical volume.

Overrides the parent LogicalVolume.create() method to automatically set type='thin-pool' since ThinPool objects are always thin pools.

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvcreate	()
**options	str	All options passed as strings to lvcreate command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

pool = ThinPool(name='pool1', vg='vg0')
pool.create(size='1G')  # type='thin-pool' is automatically set

# Can also specify additional options
pool.create(size='1G', chunksize='256k', poolmetadatasize='4M')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def create(self, *args: str, **options: str) -> bool:
    """Create Thin Pool logical volume.

    Overrides the parent LogicalVolume.create() method to automatically
    set type='thin-pool' since ThinPool objects are always thin pools.

    Args:
        *args: Additional arguments passed to lvcreate
        **options: All options passed as strings to lvcreate command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        pool.create(size='1G')  # type='thin-pool' is automatically set

        # Can also specify additional options
        pool.create(size='1G', chunksize='256k', poolmetadatasize='4M')
        ```
    """
    # Always set type to 'thin-pool' for ThinPool objects
    # This ensures the correct LV type is created regardless of what user passes
    options['type'] = 'thin-pool'

    # Call parent create method
    return super().create(*args, **options)

create_thin_pool(pool_name, vg_name, *args, **options) classmethod

Create thin pool with specified options.

Parameters:

Name	Type	Description	Default
pool_name	str	Pool name	required
vg_name	str	Volume group name	required
*args	str	Additional arguments passed to lvcreate	()
**options	str	All options passed as strings to lvcreate command	{}

Returns:

Type	Description
ThinPool	ThinPool object for the created pool

Raises:

Type	Description
AssertionError	If pool creation fails

Example

pool = ThinPool.create_thin_pool('pool1', 'vg0', size='1G')

Source code in sts_libs/src/sts/lvm/logical_volume.py

@classmethod
def create_thin_pool(cls, pool_name: str, vg_name: str, *args: str, **options: str) -> ThinPool:
    """Create thin pool with specified options.

    Args:
        pool_name: Pool name
        vg_name: Volume group name
        *args: Additional arguments passed to lvcreate
        **options: All options passed as strings to lvcreate command

    Returns:
        ThinPool object for the created pool

    Raises:
        AssertionError: If pool creation fails

    Example:
        ```python
        pool = ThinPool.create_thin_pool('pool1', 'vg0', size='1G')
        ```
    """
    pool = cls(name=pool_name, vg=vg_name)
    assert pool.create(*args, **options), f'Failed to create thin pool {pool_name}'
    return pool

create_thin_volume(lv_name, *args, **options)

Create thin volume in this pool.

Parameters:

Name	Type	Description	Default
lv_name	str	Thin volume name	required
*args	str	Additional arguments passed to lvcreate	()
**options	str	All options passed as strings to lvcreate command	{}

Returns:

Type	Description
LogicalVolume	LogicalVolume object for the created thin volume

Raises:

Type	Description
AssertionError	If volume creation fails

Example

pool = ThinPool(name='pool1', vg='vg0')
thin_lv = pool.create_thin_volume('thin1', virtualsize='500M')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def create_thin_volume(self, lv_name: str, *args: str, **options: str) -> LogicalVolume:
    """Create thin volume in this pool.

    Args:
        lv_name: Thin volume name
        *args: Additional arguments passed to lvcreate
        **options: All options passed as strings to lvcreate command

    Returns:
        LogicalVolume object for the created thin volume

    Raises:
        AssertionError: If volume creation fails

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        thin_lv = pool.create_thin_volume('thin1', virtualsize='500M')
        ```
    """
    thin_lv = LogicalVolume(name=lv_name, pool_name=self.name, vg=self.vg)
    assert thin_lv.create(*args, **options), f'Failed to create thin volume {lv_name}'
    assert self.refresh_report()

    # Add to our thin volumes list
    self.thin_volumes.append(thin_lv)

    return thin_lv

discover_thin_volumes()

Discover thin volumes in this pool.

Returns:

Type	Description
list[LogicalVolume]	List of LogicalVolume instances representing thin volumes in this pool

Source code in sts_libs/src/sts/lvm/logical_volume.py

def discover_thin_volumes(self) -> list[LogicalVolume]:
    """Discover thin volumes in this pool.

    Returns:
        List of LogicalVolume instances representing thin volumes in this pool
    """
    self.thin_volumes = []

    # Get all LVs in the VG
    all_lvs = LogicalVolume.get_all(self.vg)

    # Filter for thin volumes that belong to this pool
    for lv in all_lvs:
        if (
            lv.report
            and lv.report.pool_lv
            and lv.report.pool_lv.strip('[]') == self.name
            and lv.report.lv_layout
            and 'thin' in lv.report.lv_layout
        ):
            self.thin_volumes.append(lv)

    return self.thin_volumes

get_data_stripe_size()

Get stripe size for thin pool data component.

For thin pools, the actual stripe size information is stored in the data component (_tdata), not in the main pool LV. This method returns the stripe size from the data component.

Returns:

Type	Description
str | None	String representing stripe size, or None if not a thin pool or stripe size not found

Example

pool = ThinPool(name='pool', vg='vg0')
pool.create(stripes='2', stripesize='64k', size='100M')
stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_data_stripe_size(self) -> str | None:
    """Get stripe size for thin pool data component.

    For thin pools, the actual stripe size information is stored in the data component (_tdata),
    not in the main pool LV. This method returns the stripe size from the data component.

    Returns:
        String representing stripe size, or None if not a thin pool or stripe size not found

    Example:
        ```python
        pool = ThinPool(name='pool', vg='vg0')
        pool.create(stripes='2', stripesize='64k', size='100M')
        stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'
        ```
    """
    assert self.refresh_report()
    if self.tdata and self.tdata.report:
        return self.tdata.report.stripe_size
    return None

get_data_stripes()

Get stripe count for thin pool data component.

For thin pools, the actual stripe information is stored in the data component (_tdata), not in the main pool LV. This method returns the stripe count from the data component.

Returns:

Type	Description
str | None	String representing stripe count, or None if not a thin pool or stripes not found

Example

pool = ThinPool(name='pool', vg='vg0')
pool.create(stripes='2', size='100M')
stripe_count = pool.get_data_stripes()  # Returns '2'

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_data_stripes(self) -> str | None:
    """Get stripe count for thin pool data component.

    For thin pools, the actual stripe information is stored in the data component (_tdata),
    not in the main pool LV. This method returns the stripe count from the data component.

    Returns:
        String representing stripe count, or None if not a thin pool or stripes not found

    Example:
        ```python
        pool = ThinPool(name='pool', vg='vg0')
        pool.create(stripes='2', size='100M')
        stripe_count = pool.get_data_stripes()  # Returns '2'
        ```
    """
    assert self.refresh_report()
    if self.tdata and self.tdata.report:
        return self.tdata.report.stripes
    return None

get_pool_usage()

Get thin pool data and metadata usage percentages.

Returns:

Type	Description
tuple[str, str]	Tuple of (data_percent, metadata_percent) as strings

Example

pool = ThinPool(name='pool1', vg='vg0')
data_usage, meta_usage = pool.get_pool_usage()
print(f'Data usage: {data_usage}%, Metadata usage: {meta_usage}%')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_pool_usage(self) -> tuple[str, str]:
    """Get thin pool data and metadata usage percentages.

    Returns:
        Tuple of (data_percent, metadata_percent) as strings

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        data_usage, meta_usage = pool.get_pool_usage()
        print(f'Data usage: {data_usage}%, Metadata usage: {meta_usage}%')
        ```
    """
    if not self.vg or not self.name:
        logging.error('Volume group and logical volume name required')
        return 'unknown', 'unknown'

    # Get data usage
    data_result = self.lvs(f'{self.vg}/{self.name}', o='data_percent', noheadings='')
    data_percent = data_result.stdout.strip() if data_result.succeeded else 'unknown'

    # Get metadata usage
    meta_result = self.lvs(f'{self.vg}/{self.name}', o='metadata_percent', noheadings='')
    meta_percent = meta_result.stdout.strip() if meta_result.succeeded else 'unknown'

    return data_percent, meta_percent

get_tdata_volume()

Get the tdata (data) component of the thin pool.

The tdata volume stores the actual data for all thin volumes in the pool. It's always named _tdata by LVM convention.

Returns:

Type	Description
LogicalVolume | None	LogicalVolume instance for the tdata component, or None if not found

Example

pool = ThinPool(name='pool1', vg='vg0')
tdata = pool.get_tdata_volume()
if tdata:
    print(f'Data volume: {tdata.name}')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_tdata_volume(self) -> LogicalVolume | None:
    """Get the tdata (data) component of the thin pool.

    The tdata volume stores the actual data for all thin volumes in the pool.
    It's always named <pool_name>_tdata by LVM convention.

    Returns:
        LogicalVolume instance for the tdata component, or None if not found

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        tdata = pool.get_tdata_volume()
        if tdata:
            print(f'Data volume: {tdata.name}')
        ```
    """
    if not self.name or not self.vg:
        return None

    # LVM always names the data component as {pool_name}_tdata
    data_lv_name = f'{self.name}_tdata'

    if self.tdata:
        return self.tdata

    try:
        self.tdata = LogicalVolume(name=data_lv_name, vg=self.vg)
    except (ValueError, OSError) as e:
        logging.warning(f'Failed to get tdata volume {data_lv_name}: {e}')
        return None
    else:
        return self.tdata

get_thin_volume_count()

Get the number of thin volumes in this pool.

Returns:

Type	Description
int	Number of thin volumes in the pool

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_thin_volume_count(self) -> int:
    """Get the number of thin volumes in this pool.

    Returns:
        Number of thin volumes in the pool
    """
    if self.report and self.report.thin_count:
        try:
            return int(self.report.thin_count)
        except ValueError:
            pass

    # Fallback to counting discovered volumes
    return len(self.thin_volumes)

get_tmeta_volume()

Get the tmeta (metadata) component of the thin pool.

The tmeta volume stores the thin provisioning metadata for the pool. It's always named _tmeta by LVM convention.

Returns:

Type	Description
LogicalVolume | None	LogicalVolume instance for the tmeta component, or None if not found

Example

pool = ThinPool(name='pool1', vg='vg0')
tmeta = pool.get_tmeta_volume()
if tmeta:
    print(f'Metadata volume: {tmeta.name}')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def get_tmeta_volume(self) -> LogicalVolume | None:
    """Get the tmeta (metadata) component of the thin pool.

    The tmeta volume stores the thin provisioning metadata for the pool.
    It's always named <pool_name>_tmeta by LVM convention.

    Returns:
        LogicalVolume instance for the tmeta component, or None if not found

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        tmeta = pool.get_tmeta_volume()
        if tmeta:
            print(f'Metadata volume: {tmeta.name}')
        ```
    """
    if not self.name or not self.vg:
        return None

    # LVM always names the metadata component as {pool_name}_tmeta
    meta_lv_name = f'{self.name}_tmeta'

    if self.tmeta:
        return self.tmeta

    try:
        self.tmeta = LogicalVolume(name=meta_lv_name, vg=self.vg)
    except (ValueError, OSError) as e:
        logging.warning(f'Failed to get tmeta volume {meta_lv_name}: {e}')
        return None
    else:
        return self.tmeta

refresh_report()

Refresh Thin pool LV report data.

Creates or updates the Thin pool LV report with the latest information. Also refreshes the tdata and tmeta component reports if they exist.

Returns:

Name	Type	Description
bool	bool	True if refresh was successful

Source code in sts_libs/src/sts/lvm/logical_volume.py

def refresh_report(self) -> bool:
    """Refresh Thin pool LV report data.

    Creates or updates the Thin pool LV report with the latest information.
    Also refreshes the tdata and tmeta component reports if they exist.

    Returns:
        bool: True if refresh was successful
    """
    # Create new report if needed
    if not self.report:
        # Do not provide name and vg during init to prevent update
        self.report = LVReport()
        self.report.name = self.name
        self.report.vg = self.vg

    # Refresh the pool report
    success = self.report.refresh()

    # Refresh tdata and tmeta if they exist
    if self.get_tdata_volume() and self.tdata:
        self.tdata.refresh_report()
    if self.get_tmeta_volume() and self.tmeta:
        self.tmeta.refresh_report()

    return success

remove_thin_volumes(*, force=True)

Remove all thin volumes in this pool.

This method discovers and removes all thin volumes that belong to this pool. It should be called before removing the pool itself.

Parameters:

Name	Type	Description	Default
force	bool	If True, use force flags for removal (default: True)	True

Returns:

Type	Description
bool	True if all thin volumes were removed successfully, False otherwise

Example

pool = ThinPool(name='pool1', vg='vg0')
pool.remove_thin_volumes()  # Remove all thin volumes first
pool.remove()  # Now remove the pool

Source code in sts_libs/src/sts/lvm/logical_volume.py

def remove_thin_volumes(self, *, force: bool = True) -> bool:
    """Remove all thin volumes in this pool.

    This method discovers and removes all thin volumes that belong to this pool.
    It should be called before removing the pool itself.

    Args:
        force: If True, use force flags for removal (default: True)

    Returns:
        True if all thin volumes were removed successfully, False otherwise

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        pool.remove_thin_volumes()  # Remove all thin volumes first
        pool.remove()  # Now remove the pool
        ```
    """
    # Discover thin volumes in this pool
    self.discover_thin_volumes()

    if not self.thin_volumes:
        logging.info(f'No thin volumes found in pool {self.name}')
        return True

    all_removed = True
    for thin_lv in self.thin_volumes:
        logging.info(f'Removing thin volume {thin_lv.name} from pool {self.name}')
        success = thin_lv.remove(force='', yes='') if force else thin_lv.remove()

        if not success:
            logging.warning(f'Failed to remove thin volume {thin_lv.name}')
            all_removed = False

    # Clear the thin volumes list after removal
    if all_removed:
        self.thin_volumes = []

    return all_removed

remove_with_thin_volumes(*args, **options)

Remove thin pool along with all its thin volumes.

This is a convenience method that handles the proper cleanup order: 1. Remove all thin volumes in the pool 2. Remove the pool itself

Parameters:

Name	Type	Description	Default
*args	str	Additional arguments passed to lvremove	()
**options	str	All options passed as strings to lvremove command	{}

Returns:

Type	Description
bool	True if pool and all thin volumes were removed successfully

Example

pool = ThinPool(name='pool1', vg='vg0')
pool.create(size='1G')
pool.create_thin_volume('thin1', virtualsize='500M')
pool.remove_with_thin_volumes(force='', yes='')  # Removes everything

Source code in sts_libs/src/sts/lvm/logical_volume.py

def remove_with_thin_volumes(self, *args: str, **options: str) -> bool:
    """Remove thin pool along with all its thin volumes.

    This is a convenience method that handles the proper cleanup order:
    1. Remove all thin volumes in the pool
    2. Remove the pool itself

    Args:
        *args: Additional arguments passed to lvremove
        **options: All options passed as strings to lvremove command

    Returns:
        True if pool and all thin volumes were removed successfully

    Example:
        ```python
        pool = ThinPool(name='pool1', vg='vg0')
        pool.create(size='1G')
        pool.create_thin_volume('thin1', virtualsize='500M')
        pool.remove_with_thin_volumes(force='', yes='')  # Removes everything
        ```
    """
    # First remove all thin volumes
    if not self.remove_thin_volumes(force='force' in options or 'yes' in options):
        logging.error(f'Failed to remove thin volumes from pool {self.name}')
        return False

    # Then remove the pool
    return self.remove(*args, **options)

repair(pv_device=None, *args, **options)

Repair thin pool metadata.

Attempts to repair the thin pool's metadata using lvconvert --repair. This creates a new metadata device and attempts to recover the pool.

Parameters:

Name	Type	Description	Default
pv_device	str | None	Optional PV device to use for the new metadata	None
*args	str	Additional arguments passed to lvconvert	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Repair pool using any available space
pool.repair()

# Repair pool using specific PV for new metadata
pool.repair('/dev/sdb')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def repair(self, pv_device: str | None = None, *args: str, **options: str) -> bool:
    """Repair thin pool metadata.

    Attempts to repair the thin pool's metadata using lvconvert --repair.
    This creates a new metadata device and attempts to recover the pool.

    Args:
        pv_device: Optional PV device to use for the new metadata
        *args: Additional arguments passed to lvconvert
        **options: All options passed as strings to lvconvert command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Repair pool using any available space
        pool.repair()

        # Repair pool using specific PV for new metadata
        pool.repair('/dev/sdb')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    cmd_args = ['--repair', f'{self.vg}/{self.name}']
    if pv_device:
        cmd_args.append(pv_device)
    if args:
        cmd_args.extend(args)

    result = self._run('lvconvert', *cmd_args, **options)
    logging.info(result.stdout)
    logging.info(result.stderr)
    logging.info(result.rc)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

swap_metadata(metadata_lv, *args, **options)

Swap thin pool metadata with another LV.

Swaps the thin pool's metadata device with the specified LV. This is useful for metadata repair operations where you need to extract, check, or replace the metadata device.

The command executed is

lvconvert --thinpool / --poolmetadata

Parameters:

Name	Type	Description	Default
metadata_lv	str	The LV to swap with the pool's metadata (format: vg/lv or just lv)	required
*args	str	Additional arguments passed to lvconvert	()
**options	str	All options passed as strings to lvconvert command	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

# Create a temporary LV for metadata swap
meta_swap = LogicalVolume(name='meta_swap', vg='vg0')
meta_swap.create(size='4M')

# Swap metadata - the pool's current metadata goes to meta_swap
pool.swap_metadata('vg0/meta_swap')

# Now meta_swap contains the old metadata, pool has new empty metadata
# You can activate meta_swap and run thin_check on it

# Swap back
pool.swap_metadata('vg0/meta_swap')

Source code in sts_libs/src/sts/lvm/logical_volume.py

def swap_metadata(self, metadata_lv: str, *args: str, **options: str) -> bool:
    """Swap thin pool metadata with another LV.

    Swaps the thin pool's metadata device with the specified LV.
    This is useful for metadata repair operations where you need to
    extract, check, or replace the metadata device.

    The command executed is:
        lvconvert --thinpool <vg>/<pool> --poolmetadata <metadata_lv>

    Args:
        metadata_lv: The LV to swap with the pool's metadata (format: vg/lv or just lv)
        *args: Additional arguments passed to lvconvert
        **options: All options passed as strings to lvconvert command

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Create a temporary LV for metadata swap
        meta_swap = LogicalVolume(name='meta_swap', vg='vg0')
        meta_swap.create(size='4M')

        # Swap metadata - the pool's current metadata goes to meta_swap
        pool.swap_metadata('vg0/meta_swap')

        # Now meta_swap contains the old metadata, pool has new empty metadata
        # You can activate meta_swap and run thin_check on it

        # Swap back
        pool.swap_metadata('vg0/meta_swap')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not metadata_lv:
        logging.error('Metadata LV path required')
        return False

    # Build the command with --thinpool option
    result = self._run(
        'lvconvert',
        '--thinpool',
        f'{self.vg}/{self.name}',
        '--poolmetadata',
        metadata_lv,
        *args,
        **options,
    )
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

VolumeGroup dataclass

Bases: LvmDevice

Volume Group device.

A Volume Group (VG) combines Physical Volumes into a storage pool. This pool can then be divided into Logical Volumes.

Key features: - Combine multiple PVs - Manage storage pool - Track extent allocation - Handle PV addition/removal

Parameters:

Name	Type	Description	Default
name	str | None	Device name (optional)	None
path	Path | str | None	Device path (optional, defaults to /dev/)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
model	str | None	Device model (optional)	None
yes	bool	Automatically answer yes to prompts	True
force	bool	Force operations without confirmation	False
pvs	list[str]	List of Physical Volumes (optional, discovered from device)	list()

Example

vg = VolumeGroup(
    name='vg0'
)  # Creates Volume Group with name (this doesn't create the VG on the system, only creates the object)
vg.create(['/dev/sda1'])  # Creates VG with name 'vg0' on the system using the Physical Volume '/dev/sda1'

Source code in sts_libs/src/sts/lvm/volume_group.py

@dataclass
class VolumeGroup(LvmDevice):
    """Volume Group device.

    A Volume Group (VG) combines Physical Volumes into a storage pool.
    This pool can then be divided into Logical Volumes.

    Key features:
    - Combine multiple PVs
    - Manage storage pool
    - Track extent allocation
    - Handle PV addition/removal

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        pvs: List of Physical Volumes (optional, discovered from device)

    Example:
        ```python
        vg = VolumeGroup(
            name='vg0'
        )  # Creates Volume Group with name (this doesn't create the VG on the system, only creates the object)
        vg.create(['/dev/sda1'])  # Creates VG with name 'vg0' on the system using the Physical Volume '/dev/sda1'
        ```
    """

    # Optional parameters for this class
    pvs: list[str] = field(default_factory=list)  # Member PVs

    # Available VG commands
    COMMANDS: ClassVar[list[str]] = [
        'vgcfgbackup',  # Backup VG metadata
        'vgcfgrestore',  # Restore VG metadata
        'vgchange',  # Change VG attributes
        'vgck',  # Check VG metadata
        'vgconvert',  # Convert VG metadata format
        'vgcreate',  # Create VG
        'vgdisplay',  # Show VG details
        'vgexport',  # Make VG inactive
        'vgextend',  # Add PVs to VG
        'vgimport',  # Make VG active
        'vgimportclone',  # Import cloned PVs
        'vgimportdevices',  # Import PVs into VG
        'vgmerge',  # Merge VGs
        'vgmknodes',  # Create VG special files
        'vgreduce',  # Remove PVs from VG
        'vgremove',  # Remove VG
        'vgrename',  # Rename VG
        'vgs',  # List VGs
        'vgscan',  # Scan for VGs
        'vgsplit',  # Split VG into two
    ]

    def discover_pvs(self) -> list[str] | None:
        """Discover PVs if name is available."""
        if self.name:
            result = run(f'vgs {self.name} -o pv_name --noheadings')
            if result.succeeded:
                self.pvs = result.stdout.strip().splitlines()
                return self.pvs
        return None

    def create(self, **options: str) -> bool:
        """Create Volume Group.

        Creates a new VG from specified PVs:
        - Initializes VG metadata
        - Sets up extent allocation
        - Creates device mapper devices

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
            vg.create()
            True
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False
        if not self.pvs:
            logging.error('Physical volumes required')
            return False

        result = self._run('vgcreate', self.name, *self.pvs, **options)
        return result.succeeded

    def remove(self, **options: str) -> bool:
        """Remove Volume Group.

        Removes VG and its metadata:
        - All LVs must be removed first
        - PVs are released but not removed

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.remove()
            True
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgremove', self.name, **options)
        return result.succeeded

    def activate(self, **options: str) -> bool:
        """Activate Volume Group.

        Makes the VG and all its LVs available for use.

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.activate()
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgchange', '-a', 'y', self.name, **options)
        return result.succeeded

    def deactivate(self, **options: str) -> bool:
        """Deactivate Volume Group.

        Makes the VG and all its LVs unavailable.

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.deactivate()
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgchange', '-a', 'n', self.name, **options)
        return result.succeeded

activate(**options)

Activate Volume Group.

Makes the VG and all its LVs available for use.

Parameters:

Name	Type	Description	Default
**options	str	VG options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

vg = VolumeGroup(name='vg0')
vg.activate()

Source code in sts_libs/src/sts/lvm/volume_group.py

def activate(self, **options: str) -> bool:
    """Activate Volume Group.

    Makes the VG and all its LVs available for use.

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.activate()
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgchange', '-a', 'y', self.name, **options)
    return result.succeeded

create(**options)

Create Volume Group.

Creates a new VG from specified PVs: - Initializes VG metadata - Sets up extent allocation - Creates device mapper devices

Parameters:

Name	Type	Description	Default
**options	str	VG options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
vg.create()
True

Source code in sts_libs/src/sts/lvm/volume_group.py

def create(self, **options: str) -> bool:
    """Create Volume Group.

    Creates a new VG from specified PVs:
    - Initializes VG metadata
    - Sets up extent allocation
    - Creates device mapper devices

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
        vg.create()
        True
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False
    if not self.pvs:
        logging.error('Physical volumes required')
        return False

    result = self._run('vgcreate', self.name, *self.pvs, **options)
    return result.succeeded

deactivate(**options)

Deactivate Volume Group.

Makes the VG and all its LVs unavailable.

Parameters:

Name	Type	Description	Default
**options	str	VG options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

vg = VolumeGroup(name='vg0')
vg.deactivate()

Source code in sts_libs/src/sts/lvm/volume_group.py

def deactivate(self, **options: str) -> bool:
    """Deactivate Volume Group.

    Makes the VG and all its LVs unavailable.

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.deactivate()
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgchange', '-a', 'n', self.name, **options)
    return result.succeeded

discover_pvs()

Discover PVs if name is available.

Source code in sts_libs/src/sts/lvm/volume_group.py

def discover_pvs(self) -> list[str] | None:
    """Discover PVs if name is available."""
    if self.name:
        result = run(f'vgs {self.name} -o pv_name --noheadings')
        if result.succeeded:
            self.pvs = result.stdout.strip().splitlines()
            return self.pvs
    return None

remove(**options)

Remove Volume Group.

Removes VG and its metadata: - All LVs must be removed first - PVs are released but not removed

Parameters:

Name	Type	Description	Default
**options	str	VG options (see LvmOptions)	{}

Returns:

Type	Description
bool	True if successful, False otherwise

Example

vg = VolumeGroup(name='vg0')
vg.remove()
True

Source code in sts_libs/src/sts/lvm/volume_group.py

def remove(self, **options: str) -> bool:
    """Remove Volume Group.

    Removes VG and its metadata:
    - All LVs must be removed first
    - PVs are released but not removed

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.remove()
        True
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgremove', self.name, **options)
    return result.succeeded