tcloud
Advanced tools
| import os | ||
| def is_in_codespace() -> bool: | ||
| """ | ||
| Check if the current environment is a GitHub codespace. | ||
| """ | ||
| env_var = os.getenv("CODESPACES") | ||
| if env_var: | ||
| return env_var == "true" | ||
| return False |
+1
-1
| Metadata-Version: 2.1 | ||
| Name: tcloud | ||
| Version: 2.9.6 | ||
| Version: 2.9.7 | ||
| Author: TobikoData Inc. | ||
@@ -5,0 +5,0 @@ Author-email: engineering@tobikodata.com |
+1
-1
@@ -234,3 +234,3 @@ | ||
| 'readme': 'README.md', | ||
| 'version': '2.9.6', | ||
| 'version': '2.9.7', | ||
| }) |
| Metadata-Version: 2.1 | ||
| Name: tcloud | ||
| Version: 2.9.6 | ||
| Version: 2.9.7 | ||
| Author: TobikoData Inc. | ||
@@ -5,0 +5,0 @@ Author-email: engineering@tobikodata.com |
@@ -17,2 +17,3 @@ LICENSE | ||
| tobikodata/tcloud/auth.py | ||
| tobikodata/tcloud/codespaces.py | ||
| tobikodata/tcloud/config.py | ||
@@ -19,0 +20,0 @@ tobikodata/tcloud/constants.py |
@@ -1,1 +0,1 @@ | ||
| __version__ = version = '2.9.6' | ||
| __version__ = version = '2.9.7' |
@@ -357,3 +357,3 @@ from __future__ import annotations | ||
| def id_token(self, login: bool = False) -> t.Optional[str]: | ||
| def id_token(self, use_device_flow: bool = False, login: bool = False) -> t.Optional[str]: | ||
| """ | ||
@@ -388,3 +388,3 @@ Returns the id_token needed for SSO. Will return the one saved on disk, | ||
| # We should get a new token | ||
| return self.login() | ||
| return self.login(use_device_flow) | ||
@@ -409,18 +409,3 @@ return None | ||
| def login(self) -> t.Optional[str]: | ||
| # Can we use client credentials? | ||
| if self.client_secret: | ||
| return self.login_with_client_credentials() | ||
| server = SSOHTTPServer() | ||
| thread = threading.Thread(target=server.handle_request) | ||
| thread.start() | ||
| auth_url, _ = self.session.create_authorization_url( | ||
| self.auth_url, | ||
| code_verifier=self.code_verifier, | ||
| ) | ||
| self.console.print("Logging into [tobiko]Tobiko Cloud[/tobiko]") | ||
| def _open_browser_flow(self, auth_url: str) -> None: | ||
| try: | ||
@@ -450,2 +435,22 @@ webbrowser.open(auth_url) | ||
| def login(self, use_device_flow: bool) -> t.Optional[str]: | ||
| # Can we use client credentials? | ||
| if self.client_secret: | ||
| return self.login_with_client_credentials() | ||
| if use_device_flow: | ||
| return self.login_device_flow() | ||
| server = SSOHTTPServer() | ||
| thread = threading.Thread(target=server.handle_request) | ||
| thread.start() | ||
| auth_url, _ = self.session.create_authorization_url( | ||
| self.auth_url, | ||
| code_verifier=self.code_verifier, | ||
| ) | ||
| self.console.print("Logging into [tobiko]Tobiko Cloud[/tobiko]") | ||
| self._open_browser_flow(auth_url) | ||
| try: | ||
@@ -481,2 +486,53 @@ self.console.print() | ||
| def login_device_flow(self) -> t.Optional[str]: | ||
| response = self.session.request( | ||
| "POST", | ||
| self.auth_url + "/device", | ||
| data={"client_id": self.client_id, "scope": self.scope}, | ||
| withhold_token=True, | ||
| ) | ||
| if response.status_code != 200: | ||
| raise ValueError("Failed to initiate device flow") | ||
| response_json = response.json() | ||
| device_code = response_json["device_code"] | ||
| user_code = response_json["user_code"] | ||
| full_verification_url = response_json["verification_uri_complete"] | ||
| self.console.print("Logging into [tobiko]Tobiko Cloud[/tobiko]") | ||
| self.console.print(f"Please verify the following device code in your browser: {user_code}") | ||
| self._open_browser_flow(full_verification_url) | ||
| try: | ||
| while True: | ||
| response = self.session.request( | ||
| "POST", | ||
| self.token_url, | ||
| data={ | ||
| "grant_type": "urn:ietf:params:oauth:grant-type:device_code", | ||
| "device_code": device_code, | ||
| "client_id": self.client_id, | ||
| }, | ||
| withhold_token=True, | ||
| ) | ||
| if response.status_code == 400: | ||
| if response.json().get("error", "") == "authorization_pending": | ||
| time.sleep(5) | ||
| continue | ||
| raise ValueError("Failed to poll device flow") | ||
| if response.status_code != 200: | ||
| raise ValueError("Failed to poll device flow") | ||
| self.console.print("[success]Success![/success] :white_check_mark:") | ||
| self.console.print() | ||
| return self._create_token_info(response.json())["id_token"] | ||
| except KeyboardInterrupt: | ||
| from rich.control import Control | ||
| from rich.segment import ControlType | ||
| self.console.control(Control(ControlType.CARRIAGE_RETURN)) | ||
| self.console.print("Canceling SSO request") | ||
| self.console.print() | ||
| return None | ||
| def refresh_token(self) -> t.Optional[str]: | ||
@@ -483,0 +539,0 @@ # Can we use client credentials? |
@@ -8,2 +8,3 @@ import json | ||
| from tobikodata.tcloud.auth import tcloud_sso | ||
| from tobikodata.tcloud.codespaces import is_in_codespace | ||
@@ -104,5 +105,18 @@ SSO = tcloud_sso() | ||
| ) | ||
| def login(force: bool) -> None: | ||
| @click.option( | ||
| "-d", | ||
| "--device-flow", | ||
| is_flag=True, | ||
| default=False, | ||
| help="Use device flow to login.", | ||
| hidden=True, | ||
| ) | ||
| def login(force: bool, device_flow: bool) -> None: | ||
| """Login to Tobiko Cloud""" | ||
| SSO.login() if force else SSO.id_token(login=True) | ||
| use_device_flow = device_flow or is_in_codespace() | ||
| if force: | ||
| SSO.login(use_device_flow=use_device_flow) | ||
| else: | ||
| SSO.id_token(login=True, use_device_flow=use_device_flow) | ||
| SSO.status() | ||
@@ -109,0 +123,0 @@ |
Alert delta unavailable
Currently unable to show alert delta for PyPI packages.
102232
3.04%30
3.45%1667
4.45%