Block Device

Linux block devices (/dev/sd*, /dev/dm-*, /dev/nvme*, etc.) with sysfs-backed properties (size, sectors, scheduler, device ID) and common operations (wipefs, blkdiscard, discovery).

sts.blockdevice

Block device discovery and information (lsblk, blockdev).

BlockDevice pydantic-model

Bases: StorageDevice

Block device with lsblk/blockdev info, properties, and discovery.

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Block device with lsblk/blockdev info, properties, and discovery.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "title": "BlockDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)

Source code in sts_libs/src/sts/blockdevice.py

class BlockDevice(StorageDevice):
    """Block device with lsblk/blockdev info, properties, and discovery."""

    blockdev_info: BlockdevInfo | None = Field(default=None, init=False, repr=False)
    lsblk_info: LsblkInfo | None = Field(default=None, init=False, repr=False)

    # Sysfs path for block devices
    SYS_BLOCK_PATH: ClassVar[Path] = Path('/sys/dev/block')
    SYS_CLASS_BLOCK_PATH: ClassVar[Path] = Path('/sys/class/block')
    PROC_DISKSTATS_PATH: ClassVar[Path] = Path('/proc/diskstats')

    def discover(self) -> Self:
        """Load device data from system queries; returns self for chaining."""
        self.blockdev_info = self._load_blockdev_info()
        self.lsblk_info = self._load_lsblk_info()
        self.size = self.blockdev_info.size
        self.model = self.lsblk_info.model_name
        return self

    @classmethod
    def from_lsblk(cls, data: dict[str, Any]) -> Self:
        """Create a BlockDevice from pre-fetched lsblk data.

        Used by bulk-discovery functions that already have lsblk JSON output,
        avoiding redundant per-device system calls.
        """
        device = cls(path=Path('/dev') / data['name'])
        device.blockdev_info = BlockdevInfo.model_validate(data)
        device.lsblk_info = LsblkInfo.model_validate(data)
        device.size = device.blockdev_info.size
        device.model = device.lsblk_info.model_name
        return device

    def _load_blockdev_info(self) -> BlockdevInfo:
        """Parse ``blockdev --report`` output for this device."""
        result = run(f'blockdev --report {self.path}')
        if result.failed:
            raise DeviceError(f'Failed to get blockdev data: {result.stderr}')

        try:
            lines = result.stdout.splitlines()
            if len(lines) < MIN_BLOCKDEV_LINES:
                raise DeviceError(f'No data from {self.path}')

            header = lines[0].split()
            fields = lines[1].split()

            if header != ['RO', 'RA', 'SSZ', 'BSZ', 'StartSec', 'Size', 'Device']:
                raise DeviceError(f'Unknown output of blockdev: {header}')

            return BlockdevInfo.model_validate(
                {
                    'ro': fields[0] != 'rw',
                    'ra': int(fields[1]),
                    'log-sec': int(fields[2]),
                    'phy-sec': int(fields[3]),
                    'start': int(fields[4]),
                    'size': int(fields[5]),
                }
            )
        except (IndexError, ValueError) as e:
            raise DeviceError(f'Invalid blockdev data: {e}') from e

    def _load_lsblk_info(self) -> LsblkInfo:
        """Parse ``lsblk -JOb`` output for this device."""
        result = run(f'lsblk -JOb {self.path}')
        if result.failed:
            raise DeviceError(f'Failed to get lsblk data: {result.stderr}')

        try:
            blockdevs = json.loads(result.stdout)['blockdevices']
            if not blockdevs:
                raise DeviceError(f'No data from {self.path}')

            data = blockdevs[0]
            if 'start' not in data:
                data['start'] = self._get_start_sector(data)

        except (json.JSONDecodeError, KeyError, IndexError) as e:
            raise DeviceError(f'Invalid lsblk data: {e}') from e

        return LsblkInfo.model_validate(data)

    def _get_start_sector(self, data: dict[str, Any]) -> int:
        """Read start sector from sysfs (partitions) or return 0 (whole disks)."""
        if not data.get('pkname'):  # No parent device = whole disk
            return 0

        # Try reading start sector from sysfs
        with suppress(ValueError, DeviceError):
            result = run(f'cat {self.SYS_BLOCK_PATH}/{data["maj:min"]}/start')
            if result.succeeded and result.stdout:
                return int(result.stdout)

        return 0

    def refresh_data(self) -> None:
        """Refresh device data."""
        self.blockdev_info = self._load_blockdev_info()
        self.lsblk_info = self._load_lsblk_info()
        self.size = self.blockdev_info.size
        self.model = self.lsblk_info.model_name

    @property
    def sysfs_path(self) -> Path:
        """Return the canonical sysfs directory for this block device.

        Raises:
            DeviceError: If the device has no name.
        """
        if not self.name:
            raise DeviceError(f'Cannot determine sysfs path for device {self.path}')
        return self.SYS_CLASS_BLOCK_PATH / self.name

    @property
    def queue_path(self) -> Path:
        """Return the sysfs queue directory for this block device."""
        return self.sysfs_path / 'queue'

    def _queue_attr_path(self, attr: str) -> Path:
        """Validate and return a queue attribute path.

        Raises:
            DeviceError: If ``attr`` could escape the queue directory.
        """
        if not attr or '/' in attr or '\\' in attr or attr in {'.', '..'} or '..' in Path(attr).parts:
            raise DeviceError(f'Invalid queue attribute {attr!r} for device {self.path}')
        return self.queue_path / attr

    def get_queue_attr(self, attr: str) -> str:
        """Read a queue attribute from sysfs.

        Args:
            attr: Name of the attribute below the device's queue directory.

        Raises:
            DeviceError: If the attribute cannot be read.
        """
        path = self._queue_attr_path(attr)
        try:
            value = path.read_text().strip()
        except OSError as e:
            raise DeviceError(f'Failed to read queue attribute {attr!r} for device {self.path}: {e}') from e
        if not value:
            raise DeviceError(f'Invalid queue attribute {attr!r} for device {self.path}: empty data')
        return value

    def set_queue_attr(self, attr: str, value: str | int) -> None:
        """Write a queue attribute in sysfs.

        Args:
            attr: Name of the attribute below the device's queue directory.
            value: Value to write.

        Raises:
            DeviceError: If the attribute cannot be written.
        """
        path = self._queue_attr_path(attr)
        try:
            path.write_text(str(value))
        except OSError as e:
            raise DeviceError(f'Failed to write queue attribute {attr!r} for device {self.path}: {e}') from e

    def get_available_schedulers(self) -> list[str]:
        """Return scheduler names advertised by the kernel for this device.

        Returns:
            Scheduler names, or an empty list when the device has no scheduler attribute.

        Raises:
            DeviceError: If the scheduler attribute cannot be read.
        """
        path = self.queue_path / 'scheduler'
        try:
            contents = path.read_text()
        except FileNotFoundError:
            return []
        except OSError as e:
            raise DeviceError(f'Failed to read scheduler for device {self.path}: {e}') from e

        schedulers = contents.split()
        if not schedulers:
            raise DeviceError(f'Invalid scheduler data for device {self.path}')

        available: list[str] = []
        for scheduler in schedulers:
            has_opening_bracket = scheduler.startswith('[')
            has_closing_bracket = scheduler.endswith(']')
            if has_opening_bracket != has_closing_bracket:
                raise DeviceError(f'Invalid scheduler data for device {self.path}')
            name = scheduler[1:-1] if has_opening_bracket else scheduler
            if not name or '[' in name or ']' in name:
                raise DeviceError(f'Invalid scheduler data for device {self.path}')
            available.append(name)
        return available

    def get_scheduler(self) -> str | None:
        """Return the currently selected I/O scheduler, if supported.

        Raises:
            DeviceError: If scheduler data has no selected entry.
        """
        path = self.queue_path / 'scheduler'
        try:
            schedulers = path.read_text().split()
        except FileNotFoundError:
            return None
        except OSError as e:
            raise DeviceError(f'Failed to read scheduler for device {self.path}: {e}') from e

        current = [scheduler[1:-1] for scheduler in schedulers if scheduler.startswith('[') and scheduler.endswith(']')]
        if len(current) != 1:
            raise DeviceError(f'Invalid scheduler data for device {self.path}')
        return current[0]

    def set_scheduler(self, scheduler: str) -> None:
        """Select an I/O scheduler advertised by the device.

        Args:
            scheduler: Scheduler name to select.

        Raises:
            DeviceError: If the scheduler is unsupported or cannot be written.
        """
        if scheduler not in self.get_available_schedulers():
            raise DeviceError(f'Invalid scheduler {scheduler!r} for device {self.path}')
        try:
            (self.queue_path / 'scheduler').write_text(scheduler)
        except OSError as e:
            raise DeviceError(f'Failed to set scheduler {scheduler!r} for device {self.path}: {e}') from e

    def get_diskstats(self) -> DiskStats:
        """Read this device's stable I/O counters from ``/proc/diskstats``.

        Raises:
            DeviceError: If the device identifier or its counters are unavailable or invalid.
        """
        if not self.device_id:
            raise DeviceError(f'Cannot read diskstats for device {self.path}: missing major:minor identifier')

        try:
            lines = self.PROC_DISKSTATS_PATH.read_text().splitlines()
        except OSError as e:
            raise DeviceError(f'Failed to read diskstats for device {self.path}: {e}') from e

        for line in lines:
            fields = line.split()
            if len(fields) >= 2 and f'{fields[0]}:{fields[1]}' == self.device_id:
                if len(fields) < 14:
                    raise DeviceError(f'Invalid diskstats for device {self.path}: too few fields')
                try:
                    counters = [int(value) for value in fields[3:14]]
                except ValueError as e:
                    raise DeviceError(f'Invalid diskstats for device {self.path}: {e}') from e
                return DiskStats.model_validate(dict(zip(DiskStats.model_fields, counters, strict=True)))

        raise DeviceError(f'No diskstats entry for device {self.path} ({self.device_id})')

    def get_partitions(self) -> list[BlockDevice]:
        """Discover direct partition children using lsblk JSON.

        Raises:
            DeviceError: If lsblk cannot return valid data for this device.
        """
        result = run_argv(['lsblk', '-JOb', str(self.path)])
        if result.failed:
            raise DeviceError(f'Failed to get partitions for {self.path}: {result.stderr}')
        try:
            devices = json.loads(result.stdout)['blockdevices']
            data = devices[0]
            children = data.get('children', [])
        except (json.JSONDecodeError, KeyError, IndexError) as e:
            raise DeviceError(f'Invalid partition data for {self.path}: {e}') from e

        def is_partition_child(child: dict[str, Any]) -> bool:
            if child.get('type') == 'part':
                return True
            return (
                data.get('type') == 'loop'
                and child.get('type') == 'loop'
                and child.get('pkname') == self.name
                and bool(child.get('parttype'))
                and bool(child.get('partuuid'))
            )

        return [BlockDevice.from_lsblk(child) for child in children if is_partition_child(child)]

    def wait_for_partitions(
        self,
        expected_paths: Sequence[PathOrStr],
        *,
        timeout: float = UDEV_SETTLE_TIMEOUT,
    ) -> list[BlockDevice]:
        """Wait until lsblk reports exactly the requested direct partitions.

        Args:
            expected_paths: Complete set of partition paths expected below this device.
            timeout: Maximum time to wait in seconds.

        Raises:
            DeviceError: If the exact partition set is not observed before the timeout.
            ValueError: If the timeout or expected paths are invalid.
        """
        if not math.isfinite(timeout) or timeout < 0:
            raise ValueError(f'timeout must be a finite, non-negative number, got {timeout!r}')
        if isinstance(expected_paths, str):
            raise ValueError('expected_paths must be a sequence of paths, not a bare string')  # noqa: TRY004

        requested_paths = tuple(Path(path) for path in expected_paths)
        if len(set(requested_paths)) != len(requested_paths):
            raise ValueError('expected_paths must not contain duplicates')

        deadline = time.monotonic() + timeout
        partitions = self.get_partitions()
        while True:
            observed_paths = tuple(
                Path(partition.path) if partition.path is not None else None for partition in partitions
            )
            if len(observed_paths) == len(requested_paths) and set(observed_paths) == set(requested_paths):
                partitions_by_path = {
                    Path(partition.path): partition for partition in partitions if partition.path is not None
                }
                return [partitions_by_path[path] for path in requested_paths]
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            time.sleep(min(PARTITION_POLL_INTERVAL, remaining))
            if time.monotonic() >= deadline:
                break
            partitions = self.get_partitions()

        expected = ', '.join(str(path) for path in requested_paths)
        observed = ', '.join(str(path) for path in observed_paths)
        raise DeviceError(
            f'Timed out waiting for exact partitions on {self.path}; expected [{expected}]; last observed [{observed}]'
        )

    # Properties

    @property
    def is_partition(self) -> bool:
        """True if start sector > 0 (i.e. this is a partition, not a whole disk)."""
        return self.blockdev_info.start > 0 if self.blockdev_info else False

    @property
    def sector_size(self) -> int:
        """Return sector size for the device in bytes."""
        return self.blockdev_info.log_sec if self.blockdev_info else 0

    @property
    def block_size(self) -> int:
        """Return block size for the device in bytes."""
        return self.blockdev_info.phy_sec if self.blockdev_info else 0

    @property
    def start_sector(self) -> int:
        """Return start sector of the device on the underlying device."""
        return self.blockdev_info.start if self.blockdev_info else 0

    @property
    def is_writable(self) -> bool:
        """Return True if device is writable (no RO status)."""
        return not self.blockdev_info.ro if self.blockdev_info else True

    @property
    def ra(self) -> int:
        """Return Read Ahead for the device in 512-bytes sectors."""
        return self.blockdev_info.ra if self.blockdev_info else 0

    @property
    def nr_requests(self) -> int:
        """Return the maximum number of queued requests."""
        try:
            return int(self.get_queue_attr('nr_requests'))
        except ValueError as e:
            raise DeviceError(f'Invalid nr_requests for device {self.path}') from e

    @property
    def logical_block_size(self) -> int:
        """Return the device logical block size in bytes."""
        try:
            return int(self.get_queue_attr('logical_block_size'))
        except ValueError as e:
            raise DeviceError(f'Invalid logical_block_size for device {self.path}') from e

    @property
    def zoned(self) -> str:
        """Return the device zoned model."""
        return self.get_queue_attr('zoned')

    @property
    def is_removable(self) -> bool:
        """Return True if device is removable."""
        return self.lsblk_info.rm if self.lsblk_info else False

    @property
    def hctl(self) -> str | None:
        """Return Host:Channel:Target:Lun for SCSI devices."""
        return self.lsblk_info.hctl if self.lsblk_info else None

    @property
    def state(self) -> str | None:
        """Return state of the device."""
        return self.lsblk_info.state if self.lsblk_info else None

    @property
    def partition_type(self) -> str | None:
        """Return partition table type."""
        return self.lsblk_info.pttype if self.lsblk_info else None

    @property
    def wwn(self) -> str | None:
        """Return unique storage identifier."""
        return self.lsblk_info.wwn if self.lsblk_info else None

    @property
    def filesystem_type(self) -> str | None:
        """Return filesystem type."""
        return self.lsblk_info.fstype if self.lsblk_info else None

    @property
    def is_mounted(self) -> bool:
        """Return True if the device is mounted."""
        return bool(self.lsblk_info.mountpoint) if self.lsblk_info else False

    @property
    def type(self) -> str:
        """Return device type."""
        return self.lsblk_info.type if self.lsblk_info else 'disk'

    @property
    def transport_type(self) -> str | None:
        """Return device transport type."""
        return self.lsblk_info.tran if self.lsblk_info else None

    @property
    def device_id(self) -> str | None:
        """Return device major:minor number."""
        return self.lsblk_info.maj_min if self.lsblk_info else None

    def wipe_device(self) -> bool:
        """Remove all filesystem/raid/partition signatures via ``wipefs -a``."""
        logger.debug(f'Wiping device {self.path}')

        result = run(f'wipefs -a {self.path}')

        if result.failed:
            error_msg = f'Failed to wipe device {self.path}: {result.stderr}'
            logger.error(error_msg)
            return False

        logger.debug(f'Successfully wiped device {self.path}')
        self.refresh_data()
        return True

    def __repr__(self) -> str:
        return f'BlockDevice(path={self.path!r})'

block_size property

Return block size for the device in bytes.

device_id property

Return device major:minor number.

filesystem_type property

Return filesystem type.

hctl property

Return Host:Channel:Target:Lun for SCSI devices.

is_mounted property

Return True if the device is mounted.

is_partition property

True if start sector > 0 (i.e. this is a partition, not a whole disk).

is_removable property

Return True if device is removable.

is_writable property

Return True if device is writable (no RO status).

logical_block_size property

Return the device logical block size in bytes.

nr_requests property

Return the maximum number of queued requests.

partition_type property

Return partition table type.

queue_path property

Return the sysfs queue directory for this block device.

ra property

Return Read Ahead for the device in 512-bytes sectors.

sector_size property

Return sector size for the device in bytes.

start_sector property

Return start sector of the device on the underlying device.

state property

Return state of the device.

sysfs_path property

Return the canonical sysfs directory for this block device.

Raises:

Type	Description
DeviceError	If the device has no name.

transport_type property

Return device transport type.

type property

Return device type.

wwn property

Return unique storage identifier.

zoned property

Return the device zoned model.

discover()

Load device data from system queries; returns self for chaining.

Source code in sts_libs/src/sts/blockdevice.py

def discover(self) -> Self:
    """Load device data from system queries; returns self for chaining."""
    self.blockdev_info = self._load_blockdev_info()
    self.lsblk_info = self._load_lsblk_info()
    self.size = self.blockdev_info.size
    self.model = self.lsblk_info.model_name
    return self

from_lsblk(data) classmethod

Create a BlockDevice from pre-fetched lsblk data.

Used by bulk-discovery functions that already have lsblk JSON output, avoiding redundant per-device system calls.

Source code in sts_libs/src/sts/blockdevice.py

@classmethod
def from_lsblk(cls, data: dict[str, Any]) -> Self:
    """Create a BlockDevice from pre-fetched lsblk data.

    Used by bulk-discovery functions that already have lsblk JSON output,
    avoiding redundant per-device system calls.
    """
    device = cls(path=Path('/dev') / data['name'])
    device.blockdev_info = BlockdevInfo.model_validate(data)
    device.lsblk_info = LsblkInfo.model_validate(data)
    device.size = device.blockdev_info.size
    device.model = device.lsblk_info.model_name
    return device

get_available_schedulers()

Return scheduler names advertised by the kernel for this device.

Returns:

Type	Description
list[str]	Scheduler names, or an empty list when the device has no scheduler attribute.

Raises:

Type	Description
DeviceError	If the scheduler attribute cannot be read.

Source code in sts_libs/src/sts/blockdevice.py

def get_available_schedulers(self) -> list[str]:
    """Return scheduler names advertised by the kernel for this device.

    Returns:
        Scheduler names, or an empty list when the device has no scheduler attribute.

    Raises:
        DeviceError: If the scheduler attribute cannot be read.
    """
    path = self.queue_path / 'scheduler'
    try:
        contents = path.read_text()
    except FileNotFoundError:
        return []
    except OSError as e:
        raise DeviceError(f'Failed to read scheduler for device {self.path}: {e}') from e

    schedulers = contents.split()
    if not schedulers:
        raise DeviceError(f'Invalid scheduler data for device {self.path}')

    available: list[str] = []
    for scheduler in schedulers:
        has_opening_bracket = scheduler.startswith('[')
        has_closing_bracket = scheduler.endswith(']')
        if has_opening_bracket != has_closing_bracket:
            raise DeviceError(f'Invalid scheduler data for device {self.path}')
        name = scheduler[1:-1] if has_opening_bracket else scheduler
        if not name or '[' in name or ']' in name:
            raise DeviceError(f'Invalid scheduler data for device {self.path}')
        available.append(name)
    return available

get_diskstats()

Read this device's stable I/O counters from /proc/diskstats.

Raises:

Type	Description
DeviceError	If the device identifier or its counters are unavailable or invalid.

Source code in sts_libs/src/sts/blockdevice.py

def get_diskstats(self) -> DiskStats:
    """Read this device's stable I/O counters from ``/proc/diskstats``.

    Raises:
        DeviceError: If the device identifier or its counters are unavailable or invalid.
    """
    if not self.device_id:
        raise DeviceError(f'Cannot read diskstats for device {self.path}: missing major:minor identifier')

    try:
        lines = self.PROC_DISKSTATS_PATH.read_text().splitlines()
    except OSError as e:
        raise DeviceError(f'Failed to read diskstats for device {self.path}: {e}') from e

    for line in lines:
        fields = line.split()
        if len(fields) >= 2 and f'{fields[0]}:{fields[1]}' == self.device_id:
            if len(fields) < 14:
                raise DeviceError(f'Invalid diskstats for device {self.path}: too few fields')
            try:
                counters = [int(value) for value in fields[3:14]]
            except ValueError as e:
                raise DeviceError(f'Invalid diskstats for device {self.path}: {e}') from e
            return DiskStats.model_validate(dict(zip(DiskStats.model_fields, counters, strict=True)))

    raise DeviceError(f'No diskstats entry for device {self.path} ({self.device_id})')

get_partitions()

Discover direct partition children using lsblk JSON.

Raises:

Type	Description
DeviceError	If lsblk cannot return valid data for this device.

Source code in sts_libs/src/sts/blockdevice.py

def get_partitions(self) -> list[BlockDevice]:
    """Discover direct partition children using lsblk JSON.

    Raises:
        DeviceError: If lsblk cannot return valid data for this device.
    """
    result = run_argv(['lsblk', '-JOb', str(self.path)])
    if result.failed:
        raise DeviceError(f'Failed to get partitions for {self.path}: {result.stderr}')
    try:
        devices = json.loads(result.stdout)['blockdevices']
        data = devices[0]
        children = data.get('children', [])
    except (json.JSONDecodeError, KeyError, IndexError) as e:
        raise DeviceError(f'Invalid partition data for {self.path}: {e}') from e

    def is_partition_child(child: dict[str, Any]) -> bool:
        if child.get('type') == 'part':
            return True
        return (
            data.get('type') == 'loop'
            and child.get('type') == 'loop'
            and child.get('pkname') == self.name
            and bool(child.get('parttype'))
            and bool(child.get('partuuid'))
        )

    return [BlockDevice.from_lsblk(child) for child in children if is_partition_child(child)]

get_queue_attr(attr)

Read a queue attribute from sysfs.

Parameters:

Name	Type	Description	Default
attr	str	Name of the attribute below the device's queue directory.	required

Raises:

Type	Description
DeviceError	If the attribute cannot be read.

Source code in sts_libs/src/sts/blockdevice.py

def get_queue_attr(self, attr: str) -> str:
    """Read a queue attribute from sysfs.

    Args:
        attr: Name of the attribute below the device's queue directory.

    Raises:
        DeviceError: If the attribute cannot be read.
    """
    path = self._queue_attr_path(attr)
    try:
        value = path.read_text().strip()
    except OSError as e:
        raise DeviceError(f'Failed to read queue attribute {attr!r} for device {self.path}: {e}') from e
    if not value:
        raise DeviceError(f'Invalid queue attribute {attr!r} for device {self.path}: empty data')
    return value

get_scheduler()

Return the currently selected I/O scheduler, if supported.

Raises:

Type	Description
DeviceError	If scheduler data has no selected entry.

Source code in sts_libs/src/sts/blockdevice.py

def get_scheduler(self) -> str | None:
    """Return the currently selected I/O scheduler, if supported.

    Raises:
        DeviceError: If scheduler data has no selected entry.
    """
    path = self.queue_path / 'scheduler'
    try:
        schedulers = path.read_text().split()
    except FileNotFoundError:
        return None
    except OSError as e:
        raise DeviceError(f'Failed to read scheduler for device {self.path}: {e}') from e

    current = [scheduler[1:-1] for scheduler in schedulers if scheduler.startswith('[') and scheduler.endswith(']')]
    if len(current) != 1:
        raise DeviceError(f'Invalid scheduler data for device {self.path}')
    return current[0]

refresh_data()

Refresh device data.

Source code in sts_libs/src/sts/blockdevice.py

def refresh_data(self) -> None:
    """Refresh device data."""
    self.blockdev_info = self._load_blockdev_info()
    self.lsblk_info = self._load_lsblk_info()
    self.size = self.blockdev_info.size
    self.model = self.lsblk_info.model_name

set_queue_attr(attr, value)

Write a queue attribute in sysfs.

Parameters:

Name	Type	Description	Default
attr	str	Name of the attribute below the device's queue directory.	required
value	str | int	Value to write.	required

Raises:

Type	Description
DeviceError	If the attribute cannot be written.

Source code in sts_libs/src/sts/blockdevice.py

def set_queue_attr(self, attr: str, value: str | int) -> None:
    """Write a queue attribute in sysfs.

    Args:
        attr: Name of the attribute below the device's queue directory.
        value: Value to write.

    Raises:
        DeviceError: If the attribute cannot be written.
    """
    path = self._queue_attr_path(attr)
    try:
        path.write_text(str(value))
    except OSError as e:
        raise DeviceError(f'Failed to write queue attribute {attr!r} for device {self.path}: {e}') from e

set_scheduler(scheduler)

Select an I/O scheduler advertised by the device.

Parameters:

Name	Type	Description	Default
scheduler	str	Scheduler name to select.	required

Raises:

Type	Description
DeviceError	If the scheduler is unsupported or cannot be written.

Source code in sts_libs/src/sts/blockdevice.py

def set_scheduler(self, scheduler: str) -> None:
    """Select an I/O scheduler advertised by the device.

    Args:
        scheduler: Scheduler name to select.

    Raises:
        DeviceError: If the scheduler is unsupported or cannot be written.
    """
    if scheduler not in self.get_available_schedulers():
        raise DeviceError(f'Invalid scheduler {scheduler!r} for device {self.path}')
    try:
        (self.queue_path / 'scheduler').write_text(scheduler)
    except OSError as e:
        raise DeviceError(f'Failed to set scheduler {scheduler!r} for device {self.path}: {e}') from e

wait_for_partitions(expected_paths, *, timeout=UDEV_SETTLE_TIMEOUT)

Wait until lsblk reports exactly the requested direct partitions.

Parameters:

Name	Type	Description	Default
expected_paths	Sequence[PathOrStr]	Complete set of partition paths expected below this device.	required
timeout	float	Maximum time to wait in seconds.	UDEV_SETTLE_TIMEOUT

Raises:

Type	Description
DeviceError	If the exact partition set is not observed before the timeout.
ValueError	If the timeout or expected paths are invalid.

Source code in sts_libs/src/sts/blockdevice.py

def wait_for_partitions(
    self,
    expected_paths: Sequence[PathOrStr],
    *,
    timeout: float = UDEV_SETTLE_TIMEOUT,
) -> list[BlockDevice]:
    """Wait until lsblk reports exactly the requested direct partitions.

    Args:
        expected_paths: Complete set of partition paths expected below this device.
        timeout: Maximum time to wait in seconds.

    Raises:
        DeviceError: If the exact partition set is not observed before the timeout.
        ValueError: If the timeout or expected paths are invalid.
    """
    if not math.isfinite(timeout) or timeout < 0:
        raise ValueError(f'timeout must be a finite, non-negative number, got {timeout!r}')
    if isinstance(expected_paths, str):
        raise ValueError('expected_paths must be a sequence of paths, not a bare string')  # noqa: TRY004

    requested_paths = tuple(Path(path) for path in expected_paths)
    if len(set(requested_paths)) != len(requested_paths):
        raise ValueError('expected_paths must not contain duplicates')

    deadline = time.monotonic() + timeout
    partitions = self.get_partitions()
    while True:
        observed_paths = tuple(
            Path(partition.path) if partition.path is not None else None for partition in partitions
        )
        if len(observed_paths) == len(requested_paths) and set(observed_paths) == set(requested_paths):
            partitions_by_path = {
                Path(partition.path): partition for partition in partitions if partition.path is not None
            }
            return [partitions_by_path[path] for path in requested_paths]
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        time.sleep(min(PARTITION_POLL_INTERVAL, remaining))
        if time.monotonic() >= deadline:
            break
        partitions = self.get_partitions()

    expected = ', '.join(str(path) for path in requested_paths)
    observed = ', '.join(str(path) for path in observed_paths)
    raise DeviceError(
        f'Timed out waiting for exact partitions on {self.path}; expected [{expected}]; last observed [{observed}]'
    )

wipe_device()

Remove all filesystem/raid/partition signatures via wipefs -a.

Source code in sts_libs/src/sts/blockdevice.py

def wipe_device(self) -> bool:
    """Remove all filesystem/raid/partition signatures via ``wipefs -a``."""
    logger.debug(f'Wiping device {self.path}')

    result = run(f'wipefs -a {self.path}')

    if result.failed:
        error_msg = f'Failed to wipe device {self.path}: {result.stderr}'
        logger.error(error_msg)
        return False

    logger.debug(f'Successfully wiped device {self.path}')
    self.refresh_data()
    return True

BlockdevInfo pydantic-model

Bases: ReportModel

Parsed blockdev --report output.

Show JSON schema:

{
  "description": "Parsed ``blockdev --report`` output.",
  "properties": {
    "ro": {
      "default": false,
      "title": "Ro",
      "type": "boolean"
    },
    "ra": {
      "default": 0,
      "title": "Ra",
      "type": "integer"
    },
    "log-sec": {
      "default": 0,
      "title": "Log-Sec",
      "type": "integer"
    },
    "phy-sec": {
      "default": 0,
      "title": "Phy-Sec",
      "type": "integer"
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size": {
      "default": 0,
      "title": "Size",
      "type": "integer"
    }
  },
  "title": "BlockdevInfo",
  "type": "object"
}

Fields:

• ro (bool)
• ra (int)
• log_sec (int)
• phy_sec (int)
• start (NoneAsZero)
• size (int)

Source code in sts_libs/src/sts/blockdevice.py

class BlockdevInfo(ReportModel):
    """Parsed ``blockdev --report`` output."""

    ro: bool = False
    ra: int = 0
    log_sec: int = Field(default=0, validation_alias='log-sec')
    phy_sec: int = Field(default=0, validation_alias='phy-sec')
    start: NoneAsZero = 0
    size: int = 0

DiskStats pydantic-model

Bases: ReportModel

Stable counters reported for a block device in /proc/diskstats.

Attributes:

Name	Type	Description
reads_completed	int	Completed read operations.
reads_merged	int	Read operations merged in the block layer.
sectors_read	int	Sectors read from the device.
read_ticks	int	Milliseconds spent reading.
writes_completed	int	Completed write operations.
writes_merged	int	Write operations merged in the block layer.
sectors_written	int	Sectors written to the device.
write_ticks	int	Milliseconds spent writing.
ios_in_progress	int	I/O operations currently in progress.
io_ticks	int	Milliseconds during which I/O was in progress.
time_in_queue	int	Weighted milliseconds spent in the I/O queue.

Show JSON schema:

{
  "description": "Stable counters reported for a block device in ``/proc/diskstats``.\n\nAttributes:\n    reads_completed: Completed read operations.\n    reads_merged: Read operations merged in the block layer.\n    sectors_read: Sectors read from the device.\n    read_ticks: Milliseconds spent reading.\n    writes_completed: Completed write operations.\n    writes_merged: Write operations merged in the block layer.\n    sectors_written: Sectors written to the device.\n    write_ticks: Milliseconds spent writing.\n    ios_in_progress: I/O operations currently in progress.\n    io_ticks: Milliseconds during which I/O was in progress.\n    time_in_queue: Weighted milliseconds spent in the I/O queue.",
  "properties": {
    "reads_completed": {
      "title": "Reads Completed",
      "type": "integer"
    },
    "reads_merged": {
      "title": "Reads Merged",
      "type": "integer"
    },
    "sectors_read": {
      "title": "Sectors Read",
      "type": "integer"
    },
    "read_ticks": {
      "title": "Read Ticks",
      "type": "integer"
    },
    "writes_completed": {
      "title": "Writes Completed",
      "type": "integer"
    },
    "writes_merged": {
      "title": "Writes Merged",
      "type": "integer"
    },
    "sectors_written": {
      "title": "Sectors Written",
      "type": "integer"
    },
    "write_ticks": {
      "title": "Write Ticks",
      "type": "integer"
    },
    "ios_in_progress": {
      "title": "Ios In Progress",
      "type": "integer"
    },
    "io_ticks": {
      "title": "Io Ticks",
      "type": "integer"
    },
    "time_in_queue": {
      "title": "Time In Queue",
      "type": "integer"
    }
  },
  "required": [
    "reads_completed",
    "reads_merged",
    "sectors_read",
    "read_ticks",
    "writes_completed",
    "writes_merged",
    "sectors_written",
    "write_ticks",
    "ios_in_progress",
    "io_ticks",
    "time_in_queue"
  ],
  "title": "DiskStats",
  "type": "object"
}

Fields:

• reads_completed (int)
• reads_merged (int)
• sectors_read (int)
• read_ticks (int)
• writes_completed (int)
• writes_merged (int)
• sectors_written (int)
• write_ticks (int)
• ios_in_progress (int)
• io_ticks (int)
• time_in_queue (int)

Source code in sts_libs/src/sts/blockdevice.py

class DiskStats(ReportModel):
    """Stable counters reported for a block device in ``/proc/diskstats``.

    Attributes:
        reads_completed: Completed read operations.
        reads_merged: Read operations merged in the block layer.
        sectors_read: Sectors read from the device.
        read_ticks: Milliseconds spent reading.
        writes_completed: Completed write operations.
        writes_merged: Write operations merged in the block layer.
        sectors_written: Sectors written to the device.
        write_ticks: Milliseconds spent writing.
        ios_in_progress: I/O operations currently in progress.
        io_ticks: Milliseconds during which I/O was in progress.
        time_in_queue: Weighted milliseconds spent in the I/O queue.
    """

    reads_completed: int
    reads_merged: int
    sectors_read: int
    read_ticks: int
    writes_completed: int
    writes_merged: int
    sectors_written: int
    write_ticks: int
    ios_in_progress: int
    io_ticks: int
    time_in_queue: int

LsblkInfo pydantic-model

Bases: ReportModel

Parsed lsblk -JOb output.

Show JSON schema:

{
  "description": "Parsed ``lsblk -JOb`` output.",
  "properties": {
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "rm": {
      "default": false,
      "title": "Rm",
      "type": "boolean"
    },
    "hctl": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Hctl"
    },
    "state": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "State"
    },
    "pttype": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Pttype"
    },
    "wwn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Wwn"
    },
    "fstype": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Fstype"
    },
    "mountpoint": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Mountpoint"
    },
    "type": {
      "default": "disk",
      "title": "Type",
      "type": "string"
    },
    "tran": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Tran"
    },
    "maj:min": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Maj:Min"
    },
    "pkname": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Pkname"
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    }
  },
  "title": "LsblkInfo",
  "type": "object"
}

Config:

• protected_namespaces: ()

Fields:

• model_name (str | None)
• rm (bool)
• hctl (str | None)
• state (str | None)
• pttype (str | None)
• wwn (str | None)
• fstype (str | None)
• mountpoint (str | None)
• type (str)
• tran (str | None)
• maj_min (str | None)
• pkname (str | None)
• start (NoneAsZero)

Source code in sts_libs/src/sts/blockdevice.py

class LsblkInfo(ReportModel):
    """Parsed ``lsblk -JOb`` output."""

    model_config = ConfigDict(protected_namespaces=())

    model_name: str | None = Field(default=None, validation_alias='model')
    rm: bool = False
    hctl: str | None = None
    state: str | None = None
    pttype: str | None = None
    wwn: str | None = None
    fstype: str | None = None
    mountpoint: str | None = None
    type: str = 'disk'
    tran: str | None = None
    maj_min: str | None = Field(default=None, validation_alias='maj:min')
    pkname: str | None = None
    start: NoneAsZero = 0

filter_devices_by_block_sizes(block_devices, *, prefer_matching_block_sizes, required_devices=0)

Filter block devices by sector/block size, returning the best matching group.

Parameters:

Name	Type	Description	Default
block_devices	list[BlockDevice]	Devices to filter.	required
required_devices	int	Minimum number of devices required (defaults to env var MIN_DEVICES or 0).	0
prefer_matching_block_sizes	bool	If True, prefer devices where sector_size matches block_size.	required

Returns:

Type	Description
tuple[tuple[int, int], list[BlockDevice]]	Tuple of ((sector_size, block_size), matching_devices).

Source code in sts_libs/src/sts/blockdevice.py

def filter_devices_by_block_sizes(
    block_devices: list[BlockDevice], *, prefer_matching_block_sizes: bool, required_devices: int = 0
) -> tuple[tuple[int, int], list[BlockDevice]]:
    """Filter block devices by sector/block size, returning the best matching group.

    Args:
        block_devices: Devices to filter.
        required_devices: Minimum number of devices required (defaults to env var MIN_DEVICES or 0).
        prefer_matching_block_sizes: If True, prefer devices where sector_size matches block_size.

    Returns:
        Tuple of ((sector_size, block_size), matching_devices).
    """
    if required_devices == 0:
        required_devices = int(getenv('MIN_DEVICES', '0'))

    if not block_devices:
        logger.warning('No block devices provided')
        return (0, 0), []

    # Group devices by their block sizes
    devices_by_block_sizes: dict[tuple[int, int], list[BlockDevice]] = {}
    for disk in block_devices:
        block_sizes = (disk.sector_size, disk.block_size)
        if block_sizes in devices_by_block_sizes:
            devices_by_block_sizes[block_sizes].append(disk)
        else:
            devices_by_block_sizes[block_sizes] = [disk]

    # Find the best group of devices based on preferences
    best_group_size = (0, 0)
    best_group_devices = []
    max_devices = 0

    for block_size, devices in devices_by_block_sizes.items():
        num_devices = len(devices)

        # Skip if we don't have enough devices and block sizes don't match
        if num_devices < required_devices and prefer_matching_block_sizes and block_size[0] != block_size[1]:
            continue

        # If we prefer matching block sizes and found a matching group with enough devices
        if prefer_matching_block_sizes and block_size[0] == block_size[1] and num_devices >= required_devices:
            best_group_size = block_size
            best_group_devices = devices
            break

        # Update best group if we found more devices
        if num_devices > max_devices:
            max_devices = num_devices
            best_group_size = block_size
            best_group_devices = devices

    logger.info(
        f'Using following disks: {", ".join([str(dev.path) for dev in best_group_devices])} '
        f'with block sizes: {best_group_size}'
    )

    return best_group_size, best_group_devices

get_all()

Get all block devices on the system.

Source code in sts_libs/src/sts/blockdevice.py

def get_all() -> list[BlockDevice]:
    """Get all block devices on the system."""
    devices: list[BlockDevice] = []
    for data in _load_all_devices():
        devices.extend(_process_device_data(data))
    return devices

get_free_disks()

Get unused block devices suitable for testing.

A device is considered free if it has no parent/children, is not an LVM PV, is not mounted, is not read-only, and is at least 1 GB.

Source code in sts_libs/src/sts/blockdevice.py

def get_free_disks() -> list[BlockDevice]:
    """Get unused block devices suitable for testing.

    A device is considered free if it has no parent/children, is not an LVM PV,
    is not mounted, is not read-only, and is at least 1 GB.
    """
    free_devices: list[BlockDevice] = []
    # Get list of LVM PVs to exclude (pvs may not be installed, e.g. guests without lvm2)
    try:
        pvs = {str(pv.path) for pv in PhysicalVolume.get_all()}
    except FileNotFoundError:
        pvs: set[str] = set()

    for data in _load_all_devices():
        # Skip if device has parent (is partition) or children (has partitions)
        if data.get('pkname') or data.get('children'):
            continue

        # Skip if device is LVM PV
        if f'/dev/{data["name"]}' in pvs:
            continue

        # Skip if device is mounted
        if data.get('mountpoint'):
            continue

        # Skip if device is READ-ONLY
        # lsblk returns ro as string "0"/"1" on RHEL-8, bool on RHEL-9+
        ro = data.get('ro', False)
        if ro is True or str(ro) == '1':
            logger.debug(f'Device {data["name"]} is READ-ONLY.')
            continue

        # Skip devices that are too small to be usable for testing
        device_size = int(data.get('size') or 0)
        if device_size < MIN_USABLE_DEVICE_SIZE:
            logger.debug(f'Device {data["name"]} is too small ({device_size} bytes), skipping.')
            continue

        block_device = BlockDevice.from_lsblk(data)

        # Double check mount status (belt and suspenders)
        if not block_device.is_mounted:
            free_devices.append(block_device)

    return free_devices