Fetch the access token for the request_id from Control Plane. Args: request_id: The request ID used when the user opens the browser for authentication. Returns: The access token if it exists, None otherwise.
(request_id: str)
| 557 | |
| 558 | |
| 559 | def fetch_token(request_id: str) -> tuple[str, str]: |
| 560 | """Fetch the access token for the request_id from Control Plane. |
| 561 | |
| 562 | Args: |
| 563 | request_id: The request ID used when the user opens the browser for authentication. |
| 564 | |
| 565 | Returns: |
| 566 | The access token if it exists, None otherwise. |
| 567 | """ |
| 568 | access_token = invitation_code = "" |
| 569 | try: |
| 570 | resp = httpx.get( |
| 571 | f"{FETCH_TOKEN_ENDPOINT}/{request_id}", |
| 572 | timeout=HTTP_REQUEST_TIMEOUT, |
| 573 | ) |
| 574 | resp.raise_for_status() |
| 575 | access_token = (resp_json := resp.json()).get("access_token", "") |
| 576 | invitation_code = resp_json.get("code", "") |
| 577 | except httpx.RequestError as re: |
| 578 | console.debug(f"Unable to fetch token due to request error: {re}") |
| 579 | except httpx.HTTPError as he: |
| 580 | console.debug(f"Unable to fetch token due to {he}") |
| 581 | except json.JSONDecodeError as jde: |
| 582 | console.debug(f"Server did not respond with valid json: {jde}") |
| 583 | except KeyError as ke: |
| 584 | console.debug(f"Server response format unexpected: {ke}") |
| 585 | except Exception: |
| 586 | console.debug("Unexpected errors: {ex}") |
| 587 | |
| 588 | return access_token, invitation_code |
| 589 | |
| 590 | |
| 591 | def poll_backend(backend_url: str) -> bool: |
no test coverage detected