From df7c7f9905f2477a9d4ac6b355d28fa68dbb1245 Mon Sep 17 00:00:00 2001 From: Nitai Heeb Date: Wed, 11 Sep 2024 18:59:02 +0200 Subject: [PATCH] trying to satisfy flake8 (but it doesnt understand fstrings) --- src/hermes/commands/deposit/invenio.py | 1 - src/hermes/commands/init/base.py | 24 +++++++++++-------- .../commands/init/github_permissions.py | 4 ++-- src/hermes/commands/init/github_secrets.py | 3 +-- src/hermes/commands/init/oauth_github.py | 6 ++++- src/hermes/commands/init/oauth_process.py | 20 ++++++++-------- src/hermes/commands/init/oauth_zenodo.py | 14 +++++++++-- src/hermes/commands/init/slim_click.py | 8 +++++-- 8 files changed, 50 insertions(+), 30 deletions(-) diff --git a/src/hermes/commands/deposit/invenio.py b/src/hermes/commands/deposit/invenio.py index ae9ce9e..a9bffd0 100644 --- a/src/hermes/commands/deposit/invenio.py +++ b/src/hermes/commands/deposit/invenio.py @@ -13,7 +13,6 @@ from datetime import date, datetime from pathlib import Path from urllib.parse import urlparse -import os import requests from pydantic import BaseModel diff --git a/src/hermes/commands/init/base.py b/src/hermes/commands/init/base.py index e191737..7b438b3 100644 --- a/src/hermes/commands/init/base.py +++ b/src/hermes/commands/init/base.py @@ -52,7 +52,7 @@ def scout_current_folder() -> HermesInitFolderInfo: info.git_remote_url = whitespace_split[1] break branch_info = subprocess.run(['git', 'branch'], capture_output=True, text=True, check=True).stdout - for line in remote_info.splitlines(): + for line in branch_info.splitlines(): if line.startswith("*"): info.current_branch = line.split()[1].strip() break @@ -71,6 +71,8 @@ def wait_until_the_user_is_done(): USE_FANCY_HYPERLINKS = False + + def create_console_hyperlink(url: str, word: str) -> str: return f"\033]8;;{url}\033\\{word}\033]8;;\033\\" if USE_FANCY_HYPERLINKS else f"{word} ({url})" @@ -98,8 +100,9 @@ def __init__(self, parser: argparse.ArgumentParser): self.folder_info: HermesInitFolderInfo = HermesInitFolderInfo() def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None: - command_parser.add_argument('--only-set-refresh-token', action='store_true', default=False, dest="only_refresh_token", - help="Instead of doing the whole setup, this just stores a new refresh token as secret.") + command_parser.add_argument('--only-set-refresh-token', action='store_true', default=False, + dest="only_refresh_token", + help="Instead of the whole setup, this just stores a new refresh token as secret.") command_parser.add_argument("--github-token", action='store', default="", dest="github_token", help="Use this together with --only-set-refresh-token") @@ -129,13 +132,13 @@ def __call__(self, args: argparse.Namespace) -> None: # Abort if there is no git if not self.folder_info.has_git: sc.echo("The current directory has no `.git` subdirectory. " - "Please execute `hermes init` in the root directory of your git project.") + "Please execute `hermes init` in the root directory of your git project.") return # Abort if neither GitHub nor gitlab is used if not (self.folder_info.uses_github or self.folder_info.uses_gitlab): sc.echo("Your git project ({}) is not connected to github or gitlab. It is mandatory for HERMES to " - "use one of those hosting services.".format(self.folder_info.git_remote_url)) + "use one of those hosting services.".format(self.folder_info.git_remote_url)) return sc.echo(f"Starting to initialize HERMES in {self.folder_info.absolute_path}") @@ -159,7 +162,7 @@ def __call__(self, args: argparse.Namespace) -> None: if not self.folder_info.has_citation_cff: citation_cff_url = "https://citation-file-format.github.io/cff-initializer-javascript/#/" sc.echo("Your project does not contain a `CITATION.cff` file (yet). It would be very helpful for " - "saving important metadata which is necessary for publishing.") + "saving important metadata which is necessary for publishing.") create_cff_now = sc.confirm("Do you want to create a `CITATION.cff` file now?", default=True) if create_cff_now: sc.echo("{} to create the file. Then move it into the project folder before you continue.".format( @@ -212,7 +215,7 @@ def __call__(self, args: argparse.Namespace) -> None: # Getting Zenodo token zenodo_token = "" setup_method = sc.choose("How do you want to connect your project to your Zenodo account?", - [("o", "using OAuth (default)"), ("m", "doing it manually")], default="o") + [("o", "using OAuth (default)"), ("m", "doing it manually")], default="o") if setup_method == "o": sc.echo("Opening browser to log into your Zenodo account...") zenodo_token = oauth_zenodo.get_refresh_token() @@ -241,8 +244,10 @@ def __call__(self, args: argparse.Namespace) -> None: github_permissions.allow_actions(self.folder_info.git_remote_url, token=token) else: sc.echo("Now add this token to your {} under the name ZENODO_SANDBOX.".format( - create_console_hyperlink(self.folder_info.git_remote_url.replace(".git", "/settings/secrets/actions"), - "project's GitHub secrets") + create_console_hyperlink( + self.folder_info.git_remote_url.replace(".git", "/settings/secrets/actions"), + "project's GitHub secrets" + ) )) wait_until_the_user_is_done() sc.echo("Next go to your {} and check the checkbox which reads:".format( @@ -256,4 +261,3 @@ def __call__(self, args: argparse.Namespace) -> None: print("GITLAB INIT NOT IMPLEMENTED YET") sc.echo("HERMES was initialized.") - diff --git a/src/hermes/commands/init/github_permissions.py b/src/hermes/commands/init/github_permissions.py index 44a9cb4..f441cbb 100644 --- a/src/hermes/commands/init/github_permissions.py +++ b/src/hermes/commands/init/github_permissions.py @@ -4,9 +4,9 @@ import os import requests -import json -def allow_actions(project_url: str, token = ""): + +def allow_actions(project_url: str, token=""): # Access token obtained from GitHub OAuth process if token == "": token = os.environ.get('GITHUB_TOKEN') diff --git a/src/hermes/commands/init/github_secrets.py b/src/hermes/commands/init/github_secrets.py index e3a71ae..6bcdad6 100644 --- a/src/hermes/commands/init/github_secrets.py +++ b/src/hermes/commands/init/github_secrets.py @@ -4,7 +4,6 @@ import os import requests -import json from base64 import b64encode from nacl import encoding, public @@ -16,7 +15,7 @@ def encrypt_secret(public_key: str, secret_value: str) -> str: return b64encode(encrypted).decode("utf-8") -def create_secret(project_url: str, secret_name: str, secret_value, token = ""): +def create_secret(project_url: str, secret_name: str, secret_value, token=""): # Access token obtained from GitHub OAuth process if token == "": token = os.environ.get('GITHUB_TOKEN') diff --git a/src/hermes/commands/init/oauth_github.py b/src/hermes/commands/init/oauth_github.py index 498adfa..782529a 100644 --- a/src/hermes/commands/init/oauth_github.py +++ b/src/hermes/commands/init/oauth_github.py @@ -4,6 +4,7 @@ from hermes.commands.init.oauth_process import OauthProcess + local_port = 8333 client_id = 'Ov23ctl0gNzr9smeVIHR' client_secret = 'd516303374f7e55189fe74fb2af77f31a965ad57' @@ -12,6 +13,7 @@ redirect_uri = 'http://localhost:' + str(local_port) + '/callback' scope = "repo" + def oauth_process() -> OauthProcess: return OauthProcess( client_id=client_id, @@ -22,9 +24,11 @@ def oauth_process() -> OauthProcess: local_port=local_port ) + def get_tokens() -> dict[str: str]: """Starts the oauth procedure and returns collected tokens as dict""" return oauth_process().get_tokens() + def get_access_token() -> str: - return get_tokens().get('access_token', '') \ No newline at end of file + return get_tokens().get('access_token', '') diff --git a/src/hermes/commands/init/oauth_process.py b/src/hermes/commands/init/oauth_process.py index 2af6a28..83fd7e0 100644 --- a/src/hermes/commands/init/oauth_process.py +++ b/src/hermes/commands/init/oauth_process.py @@ -1,19 +1,18 @@ # SPDX-FileCopyrightText: 2024 Forschungszentrum Jülich # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileContributor: Nitai Heeb & David Papel +# SPDX-FileContributor: Nitai Heeb +# SPDX-FileContributor: David Pape import os -import re import threading import time import webbrowser from threading import Event - -from lxml.html.diff import token from requests_oauthlib import OAuth2Session from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import parse_qs, urlparse + class OauthProcess: def __init__(self, client_id: str, client_secret: str, authorize_url: str, token_url: str, scope: str, local_port: int = 5333): @@ -32,7 +31,7 @@ def handler(*args, **kwargs): return Handler(*args, oauth_process=self, **kwargs) return handler - def start_server(self, port : int = None): + def start_server(self, port: int = None): port = port or self.local_port with HTTPServer(("127.0.0.1", port), self.create_handler_constructor()) as server: server.serve_forever() @@ -40,21 +39,21 @@ def start_server(self, port : int = None): def kill_server(self): pass - def open_browser(self, port : int = None) -> bool: + def open_browser(self, port: int = None) -> bool: port = port or self.local_port - return webbrowser.open(f"http://localhost:{port}") + return webbrowser.open(f'http://localhost:{port}') def get_tokens_from_refresh_token(self, refresh_token: str) -> dict[str: str]: """Returns access and refresh token as dict using a refresh token""" oa_session = OAuth2Session(self.client_id, redirect_uri=self.redirect_uri, scope=self.scope) return oa_session.refresh_token(self.token_url, refresh_token=refresh_token, client_secret=self.client_secret, - include_client_id=True, client_id=self.client_id) + include_client_id=True, client_id=self.client_id) def get_tokens_from_auth_code(self, auth_code: str) -> dict[str: str]: """Returns access and refresh token as dict using an auth-code""" oa_session = OAuth2Session(self.client_id, redirect_uri=self.redirect_uri, scope=self.scope) return oa_session.fetch_token(self.token_url, client_secret=self.client_secret, include_client_id=True, - code=auth_code) + code=auth_code) def get_tokens(self) -> dict[str: str]: server_thread = threading.Thread(target=self.start_server) @@ -67,6 +66,7 @@ def get_tokens(self) -> dict[str: str]: time.sleep(.2) return self.tokens + class Handler(BaseHTTPRequestHandler): def __init__(self, *args, oauth_process: OauthProcess = None, **kwargs): self.oauth_process = oauth_process @@ -97,7 +97,7 @@ def callback(self): # Parse Query String parsed = urlparse(self.path) - #print(parse_qs(parsed.query)) + # print(parse_qs(parsed.query)) auth_code = parse_qs(parsed.query)["code"][0] tokens = self.oauth_process.get_tokens_from_auth_code(auth_code) diff --git a/src/hermes/commands/init/oauth_zenodo.py b/src/hermes/commands/init/oauth_zenodo.py index ef7e505..2023b94 100644 --- a/src/hermes/commands/init/oauth_zenodo.py +++ b/src/hermes/commands/init/oauth_zenodo.py @@ -5,8 +5,10 @@ import logging from hermes.commands.init.oauth_process import OauthProcess + USING_SANDBOX = True + local_port = 8334 redirect_uri = 'http://localhost:' + str(local_port) + '/callback' sandbox_client_id = 'QJ8Q9GBI78uOdNmVNK1Vd0oAOJHqmYGvxRxiSFxt' @@ -24,6 +26,7 @@ authorize_url = sandbox_authorize_url if USING_SANDBOX else real_authorize_url token_url = sandbox_token_url if USING_SANDBOX else real_token_url + class StringHandler(logging.Handler): def __init__(self): super().__init__() @@ -36,9 +39,13 @@ def emit(self, record): def get_logs(self): return self.log_messages + string_handler = StringHandler() + + def print_logs_for_requests(): - """Forwards all logging from the requests_oauthlib.oauth2_session module to the console so we can see the used request headers and data.""" + """Forwards all logging from the requests_oauthlib.oauth2_session module to the console so we can see the used + request headers and data.""" logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', @@ -58,6 +65,7 @@ def print_logs_for_requests(): module_logger.addHandler(string_handler) + def oauth_process() -> OauthProcess: return OauthProcess( client_id=client_id, @@ -68,9 +76,11 @@ def oauth_process() -> OauthProcess: local_port=local_port ) + def get_tokens() -> dict[str: str]: """Starts the oauth procedure and returns collected tokens as dict""" return oauth_process().get_tokens() + def get_refresh_token(with_prefix: bool = True) -> str: - return get_tokens().setdefault('refresh_token', '') \ No newline at end of file + return get_tokens().setdefault('refresh_token', '') diff --git a/src/hermes/commands/init/slim_click.py b/src/hermes/commands/init/slim_click.py index 1b6efce..1b810eb 100644 --- a/src/hermes/commands/init/slim_click.py +++ b/src/hermes/commands/init/slim_click.py @@ -8,11 +8,13 @@ PRINT_DEBUG = False + def echo(text: str, debug: bool = False): if (not debug) or PRINT_DEBUG: print(text) -def confirm(text: str, default : bool = True) -> bool: + +def confirm(text: str, default: bool = True) -> bool: while True: answer = input(text + (" [Y/n]" if default else " [y/N]") + ": ").lower() if answer == "y": @@ -24,9 +26,11 @@ def confirm(text: str, default : bool = True) -> bool: else: print("Error: invalid input") -def press_enter_to_continue(text : str = "Press ENTER to continue"): + +def press_enter_to_continue(text: str = "Press ENTER to continue"): input(text) + def choose(text: str, options: list[tuple[str, str]], default: str = "") -> str: default = default.lower() print(text)