)
)
+ @NvmeofCLICommand("nvmeof get_subsystems", model.GetSubsystems)
+ @convert_to_model(model.GetSubsystems)
+ @handle_nvmeof_error
+ def get_subsystems(self, gw_group: Optional[str] = None, traddr: Optional[str] = None):
+ return NVMeoFClient(gw_group=gw_group, traddr=traddr).stub.get_subsystems(
+ NVMeoFClient.pb2.get_subsystems_req()
+ )
+
@APIRouter("/nvmeof/subsystem/{nqn}/listener", Scope.NVME_OF)
@APIDoc("NVMe-oF Subsystem Listener Management API", "NVMe-oF Subsystem Listener")
class NVMeoFListener(RESTController):
class RequestStatus(NamedTuple):
status: Annotated[int, CliFlags.EXCLUSIVE_RESULT]
error_message: str
+
+
+class ListenAdress(NamedTuple):
+ trtype: str
+ adrfam: str
+ traddr: str
+ trsvcid: str
+ transport: Optional[str]
+ secure: Optional[bool]
+
+
+class NamespaceInfo(NamedTuple):
+ nsid: int
+ name: str
+ bdev_name: Optional[str]
+ nguid: Optional[str]
+ uuid: Optional[str]
+ anagrpid: Optional[int]
+ nonce: Optional[str]
+ auto_visible: Optional[bool]
+ hosts: List[Host]
+
+
+class SubsystemInfo(NamedTuple):
+ nqn: str
+ subtype: str
+ listen_addresses: List[ListenAdress]
+ hosts: List[Host]
+ allow_any_host: bool
+ serial_number: Optional[str]
+ model_number: Optional[str]
+ max_namespaces: Optional[int]
+ min_cntlid: Optional[int]
+ max_cntlid: Optional[int]
+ namespaces: List[Namespace]
+ has_dhchap_key: Optional[bool]
+
+
+class GetSubsystems(NamedTuple):
+ subsystems: List[Subsystem]
component="nvmeof",
)
- if response.status != 0:
+ status = getattr(response, "status", None)
+ error_message = getattr(response, "error_message", None)
+
+ if status not in (None, 0):
raise DashboardException(
- msg=response.error_message,
- code=response.status,
- http_status_code=NVMeoFError2HTTP.get(response.status, 400),
+ msg=error_message or "NVMeoF operation failed",
+ code=status,
+ http_status_code=NVMeoFError2HTTP.get(status, 400), # type: ignore[arg-type]
component="nvmeof",
)
return response