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)

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/scsi.py 
 30
 31
 32
 33
 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
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 "{self.name} $"')
            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
            self.path = Path(f'/dev/{self.name}')

        self.model = self.model_name

        return self

    @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 up_or_down_disk(self, action: str) -> bool:
        """Change the disk state.

        Args:
            action: Target state ('offline' or 'running')
        """
        state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
        try:
            # Need a newline character at the end of the action, or else it cannot be written into
            action = f'{action}\n'
            state_path.write_text(action)
        except OSError:
            logger.warning(f'Failed to write to {state_path}')
            return False
        else:
            return True

    @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 = ScsiDevice.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 
232
233
234
235
236
237
238
239
240
241
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 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 "{self.name} $"')
        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
        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 
260
261
262
263
264
265
266
267
268
269
270
271
@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 
273
274
275
276
277
278
@classmethod
def get_all_scsi_devices(cls) -> list[Self]:
    """Discover and return all SCSI devices on the system."""
    scsi_ids = ScsiDevice.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 
286
287
288
289
290
291
292
293
294
295
@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 
280
281
282
283
284
@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]

rescan_disk()

Rescan the disk to detect size or geometry changes.

Source code in sts_libs/src/sts/scsi.py 
221
222
223
224
225
226
227
228
229
230
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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

up_or_down_disk(action)

Change the disk state.

Parameters:

Name Type Description Default
action str

Target state ('offline' or 'running')

 required
Source code in sts_libs/src/sts/scsi.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def up_or_down_disk(self, action: str) -> bool:
    """Change the disk state.

    Args:
        action: Target state ('offline' or 'running')
    """
    state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
    try:
        # Need a newline character at the end of the action, or else it cannot be written into
        action = f'{action}\n'
        state_path.write_text(action)
    except OSError:
        logger.warning(f'Failed to write to {state_path}')
        return False
    else:
        return True

ScsiError

Bases: DeviceError

Base class for SCSI-related errors.

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