| 49 | return self._session_url |
| 50 | |
| 51 | def init(self) -> None: |
| 52 | error_msg: str = "Can't discover entrypoint(s)" |
| 53 | try: |
| 54 | _, _data, _ = self.client.query(endpoint=self.prefix) |
| 55 | json_data: Dict[str, Any] = json.loads(_data) |
| 56 | |
| 57 | for k, v in json_data.items(): |
| 58 | if isinstance(v, dict) and "@odata.id" in v: |
| 59 | name: str = to_snake_case(k) |
| 60 | url: str = v["@odata.id"] |
| 61 | self.log.info(f"entrypoint found: {name} = {url}") |
| 62 | self._endpoints[name] = Endpoint(url, self.client) |
| 63 | |
| 64 | try: |
| 65 | self._session_url = json_data["Links"]["Sessions"]["@odata.id"] |
| 66 | except (KeyError, TypeError): |
| 67 | self.log.warning("Session URL not found in root response") |
| 68 | self._session_url = "" |
| 69 | |
| 70 | except (URLError, KeyError, json.JSONDecodeError) as e: |
| 71 | msg = f"{error_msg}: {e}" |
| 72 | self.log.error(msg) |
| 73 | raise RuntimeError(msg) from e |
| 74 | |
| 75 | |
| 76 | class Endpoint: |