Make an RPC call to the dstack-vmm API.
(self, method: str, params: Optional[Dict] = None)
| 487 | self.client = VmmClient(base_url, auth_user, auth_password) |
| 488 | |
| 489 | def rpc_call(self, method: str, params: Optional[Dict] = None) -> Dict: |
| 490 | """Make an RPC call to the dstack-vmm API.""" |
| 491 | path = f"/prpc/{method}?json" |
| 492 | status, response = self.client.request( |
| 493 | "POST", path, headers=self.headers, body=params or {} |
| 494 | ) |
| 495 | |
| 496 | if status != 200: |
| 497 | if isinstance(response, str): |
| 498 | error_msg = response |
| 499 | else: |
| 500 | error_msg = str(response) |
| 501 | raise Exception(f"API call failed: {error_msg}") |
| 502 | |
| 503 | return response |
| 504 | |
| 505 | def list_vms(self, verbose: bool = False, json_output: bool = False) -> None: |
| 506 | """List all VMs and their status.""" |
no test coverage detected