Skip to content

Commit

Permalink
fix:jupyter notebook rich format removal (#628)
Browse files Browse the repository at this point in the history
  • Loading branch information
SafetyQuincyF authored Oct 31, 2024
1 parent 321622e commit 575f938
Show file tree
Hide file tree
Showing 3 changed files with 122 additions and 18 deletions.
10 changes: 10 additions & 0 deletions .vscode/launch.json
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,16 @@
],
"console": "integratedTerminal"
},
{
"name": "Safety Auth Login --headless",
"type": "debugpy",
"request": "launch",
"module": "safety",
"args": [
"auth","login","--headless"
],
"console": "integratedTerminal"
},
{
"name": "Safety Auth Logout",
"type": "debugpy",
Expand Down
32 changes: 14 additions & 18 deletions safety/auth/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import click
from safety.auth.cli_utils import load_auth_session

from safety.auth.utils import is_jupyter_notebook
from safety.console import main_console as console

from safety.auth.constants import AUTH_SERVER_URL, CLI_AUTH_SUCCESS, CLI_LOGOUT_SUCCESS, HOST
Expand Down Expand Up @@ -191,7 +192,6 @@ class ThreadedHTTPServer(http.server.HTTPServer):
def __init__(self, server_address: Tuple, RequestHandlerClass: Any) -> None:
"""
Initialize the ThreadedHTTPServer.
Args:
server_address (Tuple): The server address as a tuple (host, port).
RequestHandlerClass (Any): The request handler class.
Expand Down Expand Up @@ -219,9 +219,7 @@ def handle_timeout(self) -> None:
headless = kwargs.get("headless", False)
initial_state = kwargs.get("initial_state", None)
ctx = kwargs.get("ctx", None)

message = "Copy and paste this url into your browser:"

message = "Copy and paste this URL into your browser:\n⚠️ Ensure there are no extra spaces, especially at line breaks, as they may break the link."

if not headless:
# Start a threaded HTTP server to handle the callback
Expand All @@ -231,13 +229,15 @@ def handle_timeout(self) -> None:
server.ctx = ctx
server_thread = threading.Thread(target=server.handle_request)
server_thread.start()
message = f"If the browser does not automatically open in 5 seconds, " \
"copy and paste this url into your browser:"
message = f"If the browser does not automatically open in 5 seconds, copy and paste this url into your browser:"

target = uri if headless else f"{uri}&port={PORT}"
console.print(f"{message} [link={target}]{target}[/link]")
console.print()

if is_jupyter_notebook():
console.print(f"{message} {target}")
else:
console.print(f"{message} [link={target}]{target}[/link]")

if headless:
# Handle the headless mode where user manually provides the response
exchange_data = None
Expand All @@ -247,29 +247,25 @@ def handle_timeout(self) -> None:
exchange_data = json.loads(auth_code_text)
state = exchange_data["state"]
code = exchange_data["code"]
except Exception as e:
except Exception:
code = state = None

return auth_process(code=code,
state=state,
initial_state=initial_state,
code_verifier=ctx.obj.auth.code_verifier,
client=ctx.obj.auth.client)
state=state,
initial_state=initial_state,
code_verifier=ctx.obj.auth.code_verifier,
client=ctx.obj.auth.client)
else:
# Wait for the browser authentication in non-headless mode
wait_msg = "waiting for browser authentication"

with console.status(wait_msg, spinner="bouncingBar"):
time.sleep(2)
click.launch(target)
server_thread.join()

except OSError as e:
if e.errno == socket.errno.EADDRINUSE:
reason = f"The port {HOST}:{PORT} is currently being used by another" \
"application or process. Please choose a different port or " \
"terminate the conflicting application/process to free up " \
"the port."
reason = f"The port {HOST}:{PORT} is currently being used by another application or process. Please choose a different port or terminate the conflicting application/process to free up the port."
else:
reason = "An error occurred while performing this operation."

Expand Down
98 changes: 98 additions & 0 deletions safety/auth/utils.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from functools import lru_cache
import json
import logging
import sys
from typing import Any, Optional, Dict, Callable, Tuple
from authlib.integrations.requests_client import OAuth2Session
from authlib.integrations.base_client.errors import OAuthError
Expand Down Expand Up @@ -425,3 +427,99 @@ def send(self, request: requests.PreparedRequest, **kwargs: Any) -> requests.Res
"""
request.headers.pop("Authorization", None)
return super().send(request, **kwargs)


from functools import lru_cache

@lru_cache(maxsize=1)
def is_jupyter_notebook() -> bool:
"""
Detects if the code is running in a Jupyter notebook environment, including
various cloud-hosted Jupyter notebooks.
Returns:
bool: True if the environment is identified as a Jupyter notebook (or
equivalent cloud-based environment), False otherwise.
Supported environments:
- Google Colab
- Amazon SageMaker
- Azure Notebooks
- Kaggle Notebooks
- Databricks Notebooks
- Datalore by JetBrains
- Paperspace Gradient Notebooks
- Classic Jupyter Notebook and JupyterLab
Detection is done by attempting to import environment-specific packages:
- Google Colab: `google.colab`
- Amazon SageMaker: `sagemaker`
- Azure Notebooks: `azureml`
- Kaggle Notebooks: `kaggle`
- Databricks Notebooks: `dbutils`
- Datalore: `datalore`
- Paperspace Gradient: `gradient`
- Classic Jupyter: Checks if 'IPKernelApp' is in IPython config.
Example:
>>> is_jupyter_notebook()
True
"""
try:
# Detect Google Colab
import google.colab
return True
except ImportError:
pass

try:
# Detect Amazon SageMaker
import sagemaker
return True
except ImportError:
pass

try:
# Detect Azure Notebooks
import azureml
return True
except ImportError:
pass

try:
# Detect Kaggle Notebooks
import kaggle
return True
except ImportError:
pass

try:
# Detect Databricks
import dbutils
return True
except ImportError:
pass

try:
# Detect Datalore
import datalore
return True
except ImportError:
pass

try:
# Detect Paperspace Gradient Notebooks
import gradient
return True
except ImportError:
pass

try:
# Detect classic Jupyter Notebook, JupyterLab, and other IPython kernel-based environments
from IPython import get_ipython
if 'IPKernelApp' in get_ipython().config:
return True
except:
pass

return False

0 comments on commit 575f938

Please # to comment.