From 1a479dacbccc8de7ed4b98f9c4cb89f8199b9ab5 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Fri, 29 Sep 2023 17:52:06 +0100 Subject: [PATCH 1/2] Add types and docstrings. --- dab_mechanic/wikidata_oauth.py | 49 +++++++++++++++++++++------------- web_view.py | 29 +++++++++++--------- 2 files changed, 48 insertions(+), 30 deletions(-) diff --git a/dab_mechanic/wikidata_oauth.py b/dab_mechanic/wikidata_oauth.py index 5af0976..bafb54e 100644 --- a/dab_mechanic/wikidata_oauth.py +++ b/dab_mechanic/wikidata_oauth.py @@ -1,11 +1,17 @@ +"""Wikidata OAuth.""" + +import typing from urllib.parse import urlencode from flask import current_app, session +from requests.models import Response from requests_oauthlib import OAuth1Session wiki_hostname = "en.wikipedia.org" api_url = f"https://{wiki_hostname}/w/api.php" +CallParams = dict[str, str | int] + def get_edit_proxy() -> dict[str, str]: """Retrieve proxy information from config.""" @@ -16,7 +22,7 @@ def get_edit_proxy() -> dict[str, str]: return {} -def api_post_request(params: dict[str, str | int]): +def api_post_request(params: CallParams) -> Response: """HTTP Post using Oauth.""" app = current_app client_key = app.config["CLIENT_KEY"] @@ -27,11 +33,12 @@ def api_post_request(params: dict[str, str | int]): resource_owner_key=session["owner_key"], resource_owner_secret=session["owner_secret"], ) - proxies = get_edit_proxy() - return oauth.post(api_url, data=params, timeout=10, proxies=proxies) + r: Response = oauth.post(api_url, data=params, timeout=10, proxies=get_edit_proxy()) + return r -def raw_request(params): +def raw_request(params: CallParams) -> Response: + """Raw request.""" app = current_app url = api_url + "?" + urlencode(params) client_key = app.config["CLIENT_KEY"] @@ -42,43 +49,49 @@ def raw_request(params): resource_owner_key=session["owner_key"], resource_owner_secret=session["owner_secret"], ) - proxies = get_edit_proxy() - return oauth.get(url, timeout=10, proxies=proxies) + r: Response = oauth.get(url, timeout=10, proxies=get_edit_proxy()) + return r -def api_request(params): - return raw_request(params).json() +def api_request(params: CallParams) -> dict[str, typing.Any]: + """Make API request and return object parsed from JSON.""" + return typing.cast(dict[str, typing.Any], raw_request(params).json()) -def get_token(): - params = { +def get_token() -> str: + """Get csrftoken from MediaWiki API.""" + params: CallParams = { "action": "query", "meta": "tokens", "format": "json", "formatversion": 2, } reply = api_request(params) - token = reply["query"]["tokens"]["csrftoken"] + token: str = reply["query"]["tokens"]["csrftoken"] return token -def userinfo_call(): +def userinfo_call() -> dict[str, typing.Any]: """Request user information via OAuth.""" - params = {"action": "query", "meta": "userinfo", "format": "json"} + params: CallParams = {"action": "query", "meta": "userinfo", "format": "json"} return api_request(params) -def get_username(): +def get_username() -> str | None: + """Get username for current user.""" if "owner_key" not in session: - return # not authorized + return None # not authorized if "username" in session: + assert isinstance(session["username"], str) return session["username"] reply = userinfo_call() if "query" not in reply: - return - session["username"] = reply["query"]["userinfo"]["name"] + return None + username = reply["query"]["userinfo"]["name"] + assert isinstance(username, str) + session["username"] = username - return session["username"] + return username diff --git a/web_view.py b/web_view.py index 2730f23..51f341f 100755 --- a/web_view.py +++ b/web_view.py @@ -26,7 +26,7 @@ awdl_url = "https://dplbot.toolforge.org/articles_with_dab_links.php" @app.before_request -def global_user(): +def global_user() -> None: """Make username available everywhere.""" flask.g.user = wikidata_oauth.get_username() @@ -47,14 +47,15 @@ def exception_handler(e): ) -def parse_articles_with_dab_links(root: lxml.html.Element) -> list[tuple[str, int]]: +def parse_articles_with_dab_links(root: lxml.html.HtmlElement) -> list[tuple[str, int]]: """Parse Articles With Multiple Dablinks.""" articles = [] table = root.find(".//table") + assert table is not None for tr in table: title = tr[0][0].text count_text = tr[1][0].text - assert count_text.endswith(" links") + assert title and count_text and count_text.endswith(" links") count = int(count_text[:-6]) articles.append((title, count)) @@ -63,7 +64,8 @@ def parse_articles_with_dab_links(root: lxml.html.Element) -> list[tuple[str, in @app.route("/") -def index(): +def index() -> str: + """Index page.""" r = requests.get(awdl_url, params={"limit": 100}) root = lxml.html.fromstring(r.content) articles = parse_articles_with_dab_links(root) @@ -125,17 +127,17 @@ def save(enwiki: str) -> Response | str: def redirect_if_needed(enwiki: str) -> Optional[Response]: """Check if there are spaces in the article name and redirect.""" + endpoint = flask.request.endpoint + assert endpoint return ( - flask.redirect( - flask.url_for(flask.request.endpoint, enwiki=enwiki.replace(" ", "_")) - ) + flask.redirect(flask.url_for(endpoint, enwiki=enwiki.replace(" ", "_"))) if " " in enwiki else None ) @app.route("/enwiki/") -def article_page(enwiki: str) -> Response: +def article_page(enwiki: str) -> Response | str: """Article Page.""" redirect = redirect_if_needed(enwiki) if redirect: @@ -151,7 +153,8 @@ def article_page(enwiki: str) -> Response: @app.route("/oauth/start") -def start_oauth(): +def start_oauth() -> Response: + """Start OAuth.""" next_page = flask.request.args.get("next") if next_page: flask.session["after_login"] = next_page @@ -174,7 +177,8 @@ def start_oauth(): @app.route("/oauth/callback", methods=["GET"]) -def oauth_callback(): +def oauth_callback() -> Response: + """Autentication callback.""" client_key = app.config["CLIENT_KEY"] client_secret = app.config["CLIENT_SECRET"] @@ -201,11 +205,12 @@ def oauth_callback(): flask.session["owner_secret"] = oauth_tokens.get("oauth_token_secret") next_page = flask.session.get("after_login") - return flask.redirect(next_page) if next_page else flask.url_for("index") + return flask.redirect(next_page if next_page else flask.url_for("index")) @app.route("/oauth/disconnect") -def oauth_disconnect(): +def oauth_disconnect() -> Response: + """Disconnect OAuth.""" for key in "owner_key", "owner_secret", "username", "after_login": if key in flask.session: del flask.session[key] From 0ff82e1c05486b4b2f705722cd1ee98f431b76ab Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Fri, 29 Sep 2023 18:03:39 +0100 Subject: [PATCH 2/2] Update error handling to work with werkzeug 2 Closes: #1 --- templates/show_error.html | 19 ++++++------------- web_view.py | 25 +++++++++++++++++++------ 2 files changed, 25 insertions(+), 19 deletions(-) diff --git a/templates/show_error.html b/templates/show_error.html index e75e75e..dd79c3c 100644 --- a/templates/show_error.html +++ b/templates/show_error.html @@ -7,24 +7,17 @@ {% block content %}
-

Software error: {{ tb.exception_type }}

+

Software error: {{ exception_type }}

- 
{{ tb.exception }}
+ 
{{ exception }}
-{% set body %} -URL: {{ request.url }} - -{{ tb.plaintext | safe }} -{% endset %} - -

Submit as an issue on GitHub (requires an account with GitHub)

-

Traceback (most recent call last)

-{{ tb.render_summary(include_title=False) | safe }} +{{ summary | safe }} + +

Error in function "{{ last_frame.f_code.co_name }}": {{ last_frame_args | pprint }}

+
{{ last_frame.f_locals | pprint }}
-

Error in function "{{ last_frame.function_name }}": {{ last_frame_args | pprint }}

-
{{ last_frame.locals | pprint }}
{% endblock %} diff --git a/web_view.py b/web_view.py index 51f341f..5809c6e 100755 --- a/web_view.py +++ b/web_view.py @@ -3,6 +3,8 @@ import inspect import json import re +import sys +import traceback from typing import Optional import flask @@ -10,7 +12,7 @@ import lxml.html import requests import werkzeug.exceptions from requests_oauthlib import OAuth1Session -from werkzeug.debug.tbtools import get_current_traceback +from werkzeug.debug.tbtools import DebugTraceback from werkzeug.wrappers import Response from dab_mechanic import mediawiki_api, wikidata_oauth, wikipedia @@ -32,14 +34,25 @@ def global_user() -> None: @app.errorhandler(werkzeug.exceptions.InternalServerError) -def exception_handler(e): - tb = get_current_traceback() - last_frame = next(frame for frame in reversed(tb.frames) if not frame.is_library) - last_frame_args = inspect.getargs(last_frame.code) +def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str, int]: + """Handle exception.""" + exec_type, exc_value, current_traceback = sys.exc_info() + assert exc_value + tb = DebugTraceback(exc_value) + + summary = tb.render_traceback_html(include_title=False) + exc_lines = "".join(tb._te.format_exception_only()) + + last_frame = list(traceback.walk_tb(current_traceback))[-1][0] + last_frame_args = inspect.getargs(last_frame.f_code) + return ( flask.render_template( "show_error.html", - tb=tb, + plaintext=tb.render_traceback_text(), + exception=exc_lines, + exception_type=tb._te.exc_type.__name__, + summary=summary, last_frame=last_frame, last_frame_args=last_frame_args, ),