From a5e5fa20dff4d6d0c708af0dfc9bcd5cc045c47e Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Wed, 9 Oct 2019 16:09:52 +0100 Subject: [PATCH] Refactor: move some oauth code to module --- app.py | 71 ++++----------------------------------- depicts/wikidata_oauth.py | 55 ++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 65 deletions(-) create mode 100644 depicts/wikidata_oauth.py diff --git a/app.py b/app.py index 0bc6d80..7169e6e 100755 --- a/app.py +++ b/app.py @@ -3,13 +3,12 @@ from flask import Flask, render_template, url_for, redirect, request, g, jsonify, session from depicts import (utils, wdqs, commons, mediawiki, painting, saam, database, dia, rijksmuseum, npg, museodelprado, barnesfoundation, - wd_catalog, relaxed_ssl, human, wikibase) + wd_catalog, relaxed_ssl, human, wikibase, wikidata_oauth) from depicts.pager import Pagination, init_pager from depicts.model import (DepictsItem, DepictsItemAltLabel, Edit, PaintingItem, Language) from depicts.error_mail import setup_error_mail from requests_oauthlib import OAuth1Session -from urllib.parse import urlencode from werkzeug.exceptions import InternalServerError from werkzeug.debug.tbtools import get_current_traceback from sqlalchemy import func, distinct @@ -126,10 +125,10 @@ def no_existing_edit(item_id, depicts_id): @app.route('/save/Q', methods=['POST']) def save(item_id): depicts = request.form.getlist('depicts') - username = get_username() + username = wikidata_oauth.get_username() assert username - token = get_token() + token = wikidata_oauth.get_token() painting_item = PaintingItem.query.get(item_id) if painting_item is None: @@ -283,42 +282,6 @@ def oauth_disconnect(): del session[key] return random_painting() -def get_username(): - if 'owner_key' not in session: - return # not authorized - - if 'username' in session: - return session['username'] - - params = {'action': 'query', 'meta': 'userinfo', 'format': 'json'} - reply = oauth_api_request(params) - if 'query' not in reply: - return - session['username'] = reply['query']['userinfo']['name'] - - return session['username'] - -@app.route("/show_user") -def show_user(): - # Make authenticated calls to the API - params = {'action': 'query', 'meta': 'userinfo', 'format': 'json'} - reply = oauth_api_request(params)['query'] - - return repr(reply) - -def oauth_api_request(params): - url = 'https://www.wikidata.org/w/api.php?' + urlencode(params) - client_key = app.config['CLIENT_KEY'] - client_secret = app.config['CLIENT_SECRET'] - oauth = OAuth1Session(client_key, - client_secret=client_secret, - resource_owner_key=session['owner_key'], - resource_owner_secret=session['owner_secret']) - r = oauth.get(url, timeout=2) - reply = r.json() - - return reply - def create_claim(painting_id, depicts_id, token): painting_qid = f'Q{painting_id}' value = json.dumps({'entity-type': 'item', @@ -333,29 +296,7 @@ def create_claim(painting_id, depicts_id, token): 'format': 'json', 'formatversion': 2, } - return oauth_api_post_request(params) - -def get_token(): - params = { - 'action': 'query', - 'meta': 'tokens', - 'format': 'json', - 'formatversion': 2, - } - reply = oauth_api_request(params) - token = reply['query']['tokens']['csrftoken'] - - return token - -def oauth_api_post_request(params): - url = 'https://www.wikidata.org/w/api.php' - client_key = app.config['CLIENT_KEY'] - client_secret = app.config['CLIENT_SECRET'] - oauth = OAuth1Session(client_key, - client_secret=client_secret, - resource_owner_key=session['owner_key'], - resource_owner_secret=session['owner_secret']) - return oauth.post(url, data=params, timeout=2) + return wikidata_oauth.api_post_request(params) def image_with_cache(qid, image_filename, width): filename = f'cache/{qid}_{width}_image.json' @@ -556,7 +497,7 @@ def item_page(item_id): catalog_detail=catalog_detail, labels=find_more_props, entity=item.entity, - username=get_username(), + username=wikidata_oauth.get_username(), label=label, label_languages=label_languages, show_translation_links=show_translation_links, @@ -757,7 +698,7 @@ def browse_page(): if not params: return render_template('browse_index.html', props=find_more_props, - username=get_username()) + username=wikidata_oauth.get_username()) flat = '_'.join(f'{pid}={qid}' for pid, qid in params) diff --git a/depicts/wikidata_oauth.py b/depicts/wikidata_oauth.py new file mode 100644 index 0000000..bc4b7c6 --- /dev/null +++ b/depicts/wikidata_oauth.py @@ -0,0 +1,55 @@ +from flask import current_app, session +from requests_oauthlib import OAuth1Session +from urllib.parse import urlencode + +def api_post_request(params): + app = current_app() + url = 'https://www.wikidata.org/w/api.php' + client_key = app.config['CLIENT_KEY'] + client_secret = app.config['CLIENT_SECRET'] + oauth = OAuth1Session(client_key, + client_secret=client_secret, + resource_owner_key=session['owner_key'], + resource_owner_secret=session['owner_secret']) + return oauth.post(url, data=params, timeout=4) + +def api_request(params): + app = current_app() + url = 'https://www.wikidata.org/w/api.php?' + urlencode(params) + client_key = app.config['CLIENT_KEY'] + client_secret = app.config['CLIENT_SECRET'] + oauth = OAuth1Session(client_key, + client_secret=client_secret, + resource_owner_key=session['owner_key'], + resource_owner_secret=session['owner_secret']) + r = oauth.get(url, timeout=4) + reply = r.json() + + return reply + +def get_token(): + params = { + 'action': 'query', + 'meta': 'tokens', + 'format': 'json', + 'formatversion': 2, + } + reply = api_request(params) + token = reply['query']['tokens']['csrftoken'] + + return token + +def get_username(): + if 'owner_key' not in session: + return # not authorized + + if 'username' in session: + return session['username'] + + params = {'action': 'query', 'meta': 'userinfo', 'format': 'json'} + reply = api_request(params) + if 'query' not in reply: + return + session['username'] = reply['query']['userinfo']['name'] + + return session['username']