(self, result: dict[str, Any])
| 67 | return {"Authorization": f"Bearer {self.token}"} |
| 68 | |
| 69 | def wait(self, result: dict[str, Any]) -> dict[str, Any]: |
| 70 | operation_id = result["operation"]["id"] |
| 71 | while True: |
| 72 | time.sleep(10) |
| 73 | result = get_result( |
| 74 | requests.get( |
| 75 | f"{self.controlplane_api_url}/v1beta2/operations/{operation_id}", |
| 76 | headers=self.headers(), |
| 77 | ) |
| 78 | ) |
| 79 | if result["operation"]["state"] == "STATE_COMPLETED": |
| 80 | return result["operation"] |
| 81 | if result["operation"]["state"] == "STATE_FAILED": |
| 82 | raise ValueError(result) |
| 83 | if result["operation"]["state"] != "STATE_IN_PROGRESS": |
| 84 | raise ValueError(result) |
| 85 | |
| 86 | def create(self, object: str, content: dict[str, Any] | None) -> dict[str, Any]: |
| 87 | return get_result( |
nothing calls this directly
no test coverage detected