Skip to content

SCSI

SCSI device and host management via sysfs (/sys/class/scsi_device/, /sys/class/scsi_host/). Covers device scanning, host reset, and SCSI-level attributes.

sts.scsi

SCSI device discovery and operations via lsscsi and sysfs.

ScsiDevice pydantic-model

Bases: StorageDevice

SCSI device representation.

Attributes:

Name Type Description
scsi_id str | None

SCSI address in H:C:T:L format (Host:Channel:Target:LUN)
host_id str | None

Host adapter number, derived from scsi_id
Note

Construction performs no I/O. Call discover() to populate attributes that require system access (lsscsi, sysfs).

Example 
device = ScsiDevice(name='sda').discover()
device = ScsiDevice(scsi_id='0:0:0:0').discover()
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "SCSI device representation.\n\nAttributes:\n    scsi_id: SCSI address in H:C:T:L format (Host:Channel:Target:LUN)\n    host_id: Host adapter number, derived from scsi_id\n\nNote:\n    Construction performs no I/O. Call `discover()` to populate\n    attributes that require system access (lsscsi, sysfs).\n\nExample:\n    ```python\n    device = ScsiDevice(name='sda').discover()\n    device = ScsiDevice(scsi_id='0:0:0:0').discover()\n    ```",
  "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": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "scsi_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Scsi Id"
    },
    "host_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Host Id"
    }
  },
  "title": "ScsiDevice",
  "type": "object"
}

Fields:

  • name (str | None)
  • path (PathOrStr | None)
  • size (int | None)
  • model (str | None)
  • scsi_id (str | None)
  • host_id (str | None)
Source code in sts_libs/src/sts/scsi.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class ScsiDevice(StorageDevice):
    """SCSI device representation.

    Attributes:
        scsi_id: SCSI address in H:C:T:L format (Host:Channel:Target:LUN)
        host_id: Host adapter number, derived from scsi_id

    Note:
        Construction performs no I/O. Call `discover()` to populate
        attributes that require system access (lsscsi, sysfs).

    Example:
        ```python
        device = ScsiDevice(name='sda').discover()
        device = ScsiDevice(scsi_id='0:0:0:0').discover()
        ```
    """

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

    # Optional parameters for this class
    scsi_id: str | None = None  # SCSI address (H:C:T:L)
    host_id: str | None = None

    # Sysfs path for SCSI devices
    SCSI_PATH: ClassVar[Path] = Path('/sys/class/scsi_device')
    SCSI_HOST_PATH: ClassVar[Path] = Path('/sys/class/scsi_host')

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        # Field derivation only — no I/O. See discover() for system lookups.
        if self.scsi_id and not self.host_id:
            self.host_id = self.scsi_id.strip('[]').split(':')[0]
        if not self.path and self.name:
            self.path = Path(f'/dev/{self.name}')
        if self.path and not self.name:
            self.name = Path(self.path).name
        return self

    def discover(self) -> Self:
        """Populate scsi_id, name, path, and model from lsscsi/sysfs."""
        if self.name and not self.scsi_id:
            result = run(f"lsscsi | grep '/dev/{self.name}[[:space:]]*$'")
            if result.succeeded and result.stdout:
                parts = result.stdout.strip().split()
                if parts:
                    self.scsi_id = parts[0].strip('[]')
                    self.host_id = self.scsi_id.split(':')[0]
            else:
                raise ScsiError(f'SCSI device {self.name} not found via lsscsi')

        if not self.name and self.scsi_id:
            self.name = self.device_name
            if self.name:
                self.path = Path(f'/dev/{self.name}')

        self.model = self.model_name

        return self

    def rediscover(self, timeout: float = UDEV_SETTLE_TIMEOUT) -> bool:
        """Wait for this HCTL's current block node and refresh its identity.

        Args:
            timeout: Maximum time to wait in seconds.

        Raises:
            ScsiError: If the HCTL maps to multiple block devices.
            ValueError: If the timeout is negative or not finite.
        """
        if not math.isfinite(timeout) or timeout < 0:
            raise ValueError(f'timeout must be a finite, non-negative number, got {timeout!r}')
        if not self.scsi_id:
            return False

        block_path = self.SCSI_PATH / self.scsi_id / 'device/block'
        deadline = time.monotonic() + timeout
        while True:
            try:
                names = sorted(mapping.name for mapping in block_path.iterdir())
            except OSError:
                names = []

            if len(names) > 1:
                mappings = ', '.join(names)
                raise ScsiError(f'Ambiguous block mappings for SCSI device {self.scsi_id}: {mappings}')
            if names:
                candidate_name = names[0]
                candidate_path = self.DEV_PATH / candidate_name
                if candidate_path.exists():
                    # Commit only after all potentially failing discovery I/O succeeds.
                    self.name, self.path = candidate_name, candidate_path
                    return True

            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            time.sleep(min(SCSI_REDISCOVERY_POLL_INTERVAL, remaining))
            if time.monotonic() >= deadline:
                break

        logger.warning(f'Timeout rediscovering block node for SCSI device {self.scsi_id}')
        return False

    @property
    def vendor(self) -> str | None:
        """Device vendor string from sysfs (e.g., 'ATA', 'NETAPP')."""
        if not self.scsi_id:
            return None

        try:
            vendor_path = self.SCSI_PATH / self.scsi_id / 'device/vendor'
            return vendor_path.read_text().strip()
        except OSError:
            return None

    @property
    def model_name(self) -> str | None:
        """Device model name from sysfs."""
        if not self.scsi_id:
            return None

        try:
            model_path = self.SCSI_PATH / self.scsi_id / 'device/model'
            return model_path.read_text().strip()
        except OSError:
            return None

    @property
    def revision(self) -> str | None:
        """Device firmware revision from sysfs."""
        if not self.scsi_id:
            return None

        try:
            rev_path = self.SCSI_PATH / self.scsi_id / 'device/rev'
            return rev_path.read_text().strip()
        except OSError:
            return None

    @property
    def device_name(self) -> str | None:
        """Block device name from sysfs (e.g., 'sdc')."""
        if not self.scsi_id:
            return None

        try:
            block_path = Path(f'{self.SCSI_PATH}/{self.scsi_id}/device/block')
            if block_path.exists():
                return next(block_path.iterdir()).name
        except OSError:
            logger.warning(f'Failed to get the device name for {self.scsi_id}')
            return None
        else:
            return None

    @property
    def transport(self) -> str | None:
        """Device transport type from lsscsi (e.g., 'iSCSI', 'fc:')."""
        if not self.scsi_id:
            return None

        try:
            result = run(f'lsscsi {self.scsi_id} --list -t | grep transport')
            if result.succeeded:
                for line in result.stdout.splitlines():
                    if 'transport' in line:
                        return line.split('=')[1].strip()
                    return None
        except (OSError, IndexError):
            return None

    @property
    def driver(self) -> str | None:
        """Host adapter driver name from lsscsi (e.g., 'qla2xxx')."""
        if not self.host_id:
            return None

        try:
            result = run(f'lsscsi -H {self.host_id}')
            if result.succeeded and len(result.stdout.split()) > 1:
                return result.stdout.split()[1]
        except (OSError, IndexError):
            logger.warning(f'Failed to get the driver for host {self.host_id}.')
            return None
        else:
            return None

    @property
    def state(self) -> str | None:
        """Device state from sysfs (e.g., 'running', 'offline')."""
        state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
        try:
            return state_path.read_text().strip()
        except OSError:
            logger.warning(f'Failed to read {state_path}')
            return None

    @property
    def pci_id(self) -> str | None:
        """PCI ID of the SCSI host adapter (e.g., '0000:08:00.0')."""
        if not self.host_id:
            return None

        scsi_host_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}')
        try:
            link = scsi_host_path.resolve().as_posix()
        except OSError as e:
            logger.warning(f'Error resolving path for host{self.host_id}: {e}')
            return None
        else:
            regex_pci_id = r'([0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f])'
            pci_match = re.search(f'{regex_pci_id}/host{self.host_id}/scsi_host', link)
            return pci_match.group(1) if pci_match else None

    def rescan_host(self) -> bool:
        """Rescan the SCSI host adapter by writing '- - -' to its scan file.

        Scans all channels, targets, and LUNs to detect new devices.
        """
        if not self.host_id:
            return False

        rescan_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}/scan')
        try:
            rescan_path.write_text('- - -')
        except OSError:
            logger.warning(f'Failed to write to {rescan_path}.')
            return False
        else:
            return True

    def rescan_disk(self) -> bool:
        """Rescan the disk to detect size or geometry changes."""
        rescan_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/rescan')
        try:
            rescan_path.write_text('1')
        except OSError:
            logger.warning(f'Failed to write to {rescan_path}.')
            return False
        else:
            return True

    def delete_disk(self) -> bool:
        """Delete the disk by writing to its sysfs delete file."""
        del_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/delete')
        try:
            del_path.write_text('1')
        except OSError:
            logger.warning(f'Failed to write to {del_path}.')
            return False
        else:
            return True

    def set_state(self, state: str) -> bool:
        """Change the disk state.

        Args:
            state: Target state ('offline' or 'running')
        """
        state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
        try:
            state_path.write_text(f'{state}\n')
        except OSError:
            logger.warning(f'Failed to write to {state_path}')
            return False
        else:
            return True

    def up_or_down_disk(self, action: str) -> bool:
        """Change disk state; compatibility alias for :meth:`set_state`."""
        return self.set_state(action)

    @classmethod
    def get_all_scsi_device_ids(cls) -> list[str]:
        """Get all SCSI device IDs from sysfs."""
        try:
            path = cls.SCSI_PATH
            if path.exists():
                return [d.name for d in path.iterdir()]
        except OSError:
            logger.warning(f'Failed to list SCSI devices from {cls.SCSI_PATH}')
            return []
        else:
            return []

    @classmethod
    def get_all_scsi_devices(cls) -> list[Self]:
        """Discover and return all SCSI devices on the system."""
        scsi_ids = cls.get_all_scsi_device_ids()
        devices: list[Self] = [cls(scsi_id=scsi_id).discover() for scsi_id in scsi_ids]
        return devices

    @classmethod
    def get_by_vendor(cls, vendor: str) -> list[Self]:
        """Get all SCSI devices matching the given vendor string."""
        all_devices = cls.get_all_scsi_devices()
        return [device for device in all_devices if device.vendor == vendor]

    @classmethod
    def get_by_attribute(cls, attribute: str, value: str) -> list[Self]:
        """Get all SCSI devices where the given attribute equals value.

        Args:
            attribute: Property name to filter on (e.g., 'transport', 'vendor')
            value: Expected value (e.g., 'fc:', 'NETAPP')
        """
        all_devices = cls.get_all_scsi_devices()
        return [device for device in all_devices if getattr(device, attribute, None) == value]

device_name property

Block device name from sysfs (e.g., 'sdc').

driver property

Host adapter driver name from lsscsi (e.g., 'qla2xxx').

model_name property

Device model name from sysfs.

pci_id property

PCI ID of the SCSI host adapter (e.g., '0000:08:00.0').

revision property

Device firmware revision from sysfs.

state property

Device state from sysfs (e.g., 'running', 'offline').

transport property

Device transport type from lsscsi (e.g., 'iSCSI', 'fc:').

vendor property

Device vendor string from sysfs (e.g., 'ATA', 'NETAPP').

delete_disk()

Delete the disk by writing to its sysfs delete file.

Source code in sts_libs/src/sts/scsi.py 
281
282
283
284
285
286
287
288
289
290
def delete_disk(self) -> bool:
    """Delete the disk by writing to its sysfs delete file."""
    del_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/delete')
    try:
        del_path.write_text('1')
    except OSError:
        logger.warning(f'Failed to write to {del_path}.')
        return False
    else:
        return True

discover()

Populate scsi_id, name, path, and model from lsscsi/sysfs.

Source code in sts_libs/src/sts/scsi.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def discover(self) -> Self:
    """Populate scsi_id, name, path, and model from lsscsi/sysfs."""
    if self.name and not self.scsi_id:
        result = run(f"lsscsi | grep '/dev/{self.name}[[:space:]]*$'")
        if result.succeeded and result.stdout:
            parts = result.stdout.strip().split()
            if parts:
                self.scsi_id = parts[0].strip('[]')
                self.host_id = self.scsi_id.split(':')[0]
        else:
            raise ScsiError(f'SCSI device {self.name} not found via lsscsi')

    if not self.name and self.scsi_id:
        self.name = self.device_name
        if self.name:
            self.path = Path(f'/dev/{self.name}')

    self.model = self.model_name

    return self

get_all_scsi_device_ids() classmethod

Get all SCSI device IDs from sysfs.

Source code in sts_libs/src/sts/scsi.py 
311
312
313
314
315
316
317
318
319
320
321
322
@classmethod
def get_all_scsi_device_ids(cls) -> list[str]:
    """Get all SCSI device IDs from sysfs."""
    try:
        path = cls.SCSI_PATH
        if path.exists():
            return [d.name for d in path.iterdir()]
    except OSError:
        logger.warning(f'Failed to list SCSI devices from {cls.SCSI_PATH}')
        return []
    else:
        return []

get_all_scsi_devices() classmethod

Discover and return all SCSI devices on the system.

Source code in sts_libs/src/sts/scsi.py 
324
325
326
327
328
329
@classmethod
def get_all_scsi_devices(cls) -> list[Self]:
    """Discover and return all SCSI devices on the system."""
    scsi_ids = cls.get_all_scsi_device_ids()
    devices: list[Self] = [cls(scsi_id=scsi_id).discover() for scsi_id in scsi_ids]
    return devices

get_by_attribute(attribute, value) classmethod

Get all SCSI devices where the given attribute equals value.

Parameters:

Name Type Description Default
attribute str

Property name to filter on (e.g., 'transport', 'vendor')

 required
value str

Expected value (e.g., 'fc:', 'NETAPP')

 required
Source code in sts_libs/src/sts/scsi.py 
337
338
339
340
341
342
343
344
345
346
@classmethod
def get_by_attribute(cls, attribute: str, value: str) -> list[Self]:
    """Get all SCSI devices where the given attribute equals value.

    Args:
        attribute: Property name to filter on (e.g., 'transport', 'vendor')
        value: Expected value (e.g., 'fc:', 'NETAPP')
    """
    all_devices = cls.get_all_scsi_devices()
    return [device for device in all_devices if getattr(device, attribute, None) == value]

get_by_vendor(vendor) classmethod

Get all SCSI devices matching the given vendor string.

Source code in sts_libs/src/sts/scsi.py 
331
332
333
334
335
@classmethod
def get_by_vendor(cls, vendor: str) -> list[Self]:
    """Get all SCSI devices matching the given vendor string."""
    all_devices = cls.get_all_scsi_devices()
    return [device for device in all_devices if device.vendor == vendor]

rediscover(timeout=UDEV_SETTLE_TIMEOUT)

Wait for this HCTL's current block node and refresh its identity.

Parameters:

Name Type Description Default
timeout float

Maximum time to wait in seconds.

 UDEV_SETTLE_TIMEOUT

Raises:

Type Description
ScsiError

If the HCTL maps to multiple block devices.
ValueError

If the timeout is negative or not finite.
Source code in sts_libs/src/sts/scsi.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def rediscover(self, timeout: float = UDEV_SETTLE_TIMEOUT) -> bool:
    """Wait for this HCTL's current block node and refresh its identity.

    Args:
        timeout: Maximum time to wait in seconds.

    Raises:
        ScsiError: If the HCTL maps to multiple block devices.
        ValueError: If the timeout is negative or not finite.
    """
    if not math.isfinite(timeout) or timeout < 0:
        raise ValueError(f'timeout must be a finite, non-negative number, got {timeout!r}')
    if not self.scsi_id:
        return False

    block_path = self.SCSI_PATH / self.scsi_id / 'device/block'
    deadline = time.monotonic() + timeout
    while True:
        try:
            names = sorted(mapping.name for mapping in block_path.iterdir())
        except OSError:
            names = []

        if len(names) > 1:
            mappings = ', '.join(names)
            raise ScsiError(f'Ambiguous block mappings for SCSI device {self.scsi_id}: {mappings}')
        if names:
            candidate_name = names[0]
            candidate_path = self.DEV_PATH / candidate_name
            if candidate_path.exists():
                # Commit only after all potentially failing discovery I/O succeeds.
                self.name, self.path = candidate_name, candidate_path
                return True

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        time.sleep(min(SCSI_REDISCOVERY_POLL_INTERVAL, remaining))
        if time.monotonic() >= deadline:
            break

    logger.warning(f'Timeout rediscovering block node for SCSI device {self.scsi_id}')
    return False

rescan_disk()

Rescan the disk to detect size or geometry changes.

Source code in sts_libs/src/sts/scsi.py 
270
271
272
273
274
275
276
277
278
279
def rescan_disk(self) -> bool:
    """Rescan the disk to detect size or geometry changes."""
    rescan_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/rescan')
    try:
        rescan_path.write_text('1')
    except OSError:
        logger.warning(f'Failed to write to {rescan_path}.')
        return False
    else:
        return True

rescan_host()

Rescan the SCSI host adapter by writing '- - -' to its scan file.

Scans all channels, targets, and LUNs to detect new devices.

Source code in sts_libs/src/sts/scsi.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def rescan_host(self) -> bool:
    """Rescan the SCSI host adapter by writing '- - -' to its scan file.

    Scans all channels, targets, and LUNs to detect new devices.
    """
    if not self.host_id:
        return False

    rescan_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}/scan')
    try:
        rescan_path.write_text('- - -')
    except OSError:
        logger.warning(f'Failed to write to {rescan_path}.')
        return False
    else:
        return True

set_state(state)

Change the disk state.

Parameters:

Name Type Description Default
state str

Target state ('offline' or 'running')

 required
Source code in sts_libs/src/sts/scsi.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def set_state(self, state: str) -> bool:
    """Change the disk state.

    Args:
        state: Target state ('offline' or 'running')
    """
    state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
    try:
        state_path.write_text(f'{state}\n')
    except OSError:
        logger.warning(f'Failed to write to {state_path}')
        return False
    else:
        return True

up_or_down_disk(action)

Change disk state; compatibility alias for :meth:set_state.

Source code in sts_libs/src/sts/scsi.py 
307
308
309
def up_or_down_disk(self, action: str) -> bool:
    """Change disk state; compatibility alias for :meth:`set_state`."""
    return self.set_state(action)

ScsiError

Bases: DeviceError

Base class for SCSI-related errors.

Source code in sts_libs/src/sts/scsi.py 
30
31
class ScsiError(DeviceError):
    """Base class for SCSI-related errors."""