Skip to content

Commit

Permalink
trying to satisfy flake8 (but it doesnt understand fstrings)
Browse files Browse the repository at this point in the history
  • Loading branch information
nheeb committed Sep 11, 2024
1 parent 6c23751 commit df7c7f9
Show file tree
Hide file tree
Showing 8 changed files with 50 additions and 30 deletions.
1 change: 0 additions & 1 deletion src/hermes/commands/deposit/invenio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
24 changes: 14 additions & 10 deletions src/hermes/commands/init/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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})"

Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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}")
Expand All @@ -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(
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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(
Expand All @@ -256,4 +261,3 @@ def __call__(self, args: argparse.Namespace) -> None:
print("GITLAB INIT NOT IMPLEMENTED YET")

sc.echo("HERMES was initialized.")

4 changes: 2 additions & 2 deletions src/hermes/commands/init/github_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down
3 changes: 1 addition & 2 deletions src/hermes/commands/init/github_secrets.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@

import os
import requests
import json
from base64 import b64encode
from nacl import encoding, public

Expand All @@ -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')
Expand Down
6 changes: 5 additions & 1 deletion src/hermes/commands/init/oauth_github.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from hermes.commands.init.oauth_process import OauthProcess


local_port = 8333
client_id = 'Ov23ctl0gNzr9smeVIHR'
client_secret = 'd516303374f7e55189fe74fb2af77f31a965ad57'
Expand All @@ -12,6 +13,7 @@
redirect_uri = 'http://localhost:' + str(local_port) + '/callback'
scope = "repo"


def oauth_process() -> OauthProcess:
return OauthProcess(
client_id=client_id,
Expand All @@ -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', '')
return get_tokens().get('access_token', '')
20 changes: 10 additions & 10 deletions src/hermes/commands/init/oauth_process.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -32,29 +31,29 @@ 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()

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)
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand Down
14 changes: 12 additions & 2 deletions src/hermes/commands/init/oauth_zenodo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand All @@ -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__()
Expand All @@ -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',
Expand All @@ -58,6 +65,7 @@ def print_logs_for_requests():

module_logger.addHandler(string_handler)


def oauth_process() -> OauthProcess:
return OauthProcess(
client_id=client_id,
Expand All @@ -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', '')
return get_tokens().setdefault('refresh_token', '')
8 changes: 6 additions & 2 deletions src/hermes/commands/init/slim_click.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand All @@ -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)
Expand Down

0 comments on commit df7c7f9

Please sign in to comment.