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)

Validators:

• _derive_fields

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')

    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

    # 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 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).

partition_type property

Return partition table type.

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.

transport_type property

Return device transport type.

type property

Return device type.

wwn property

Return unique storage identifier.

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

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

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

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 = {str(pv.path) for pv in PhysicalVolume.get_all()}

    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