aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorBryan Newbold <bnewbold@robocracy.org>2021-11-03 15:45:35 -0700
committerBryan Newbold <bnewbold@robocracy.org>2021-11-03 16:52:44 -0700
commit91bd35a1ee0993126c369e39fbf8f81f775840ee (patch)
tree87e8bc6257359326f26102e7765e508d5529f0f5
parent2e92e28df34d302fe02d1e1ff7169b7888648b9c (diff)
downloadfatcat-91bd35a1ee0993126c369e39fbf8f81f775840ee.tar.gz
fatcat-91bd35a1ee0993126c369e39fbf8f81f775840ee.zip
web: add type annotations
This commit does not include type fixes, only annotations. A small number of tuples were also converted to lists.
-rw-r--r--python/fatcat_web/__init__.py2
-rw-r--r--python/fatcat_web/auth.py15
-rw-r--r--python/fatcat_web/cors.py21
-rw-r--r--python/fatcat_web/editing_routes.py163
-rw-r--r--python/fatcat_web/entity_helpers.py30
-rw-r--r--python/fatcat_web/forms.py61
-rw-r--r--python/fatcat_web/graphics.py2
-rw-r--r--python/fatcat_web/hacks.py8
-rw-r--r--python/fatcat_web/kafka.py8
-rw-r--r--python/fatcat_web/ref_routes.py29
-rw-r--r--python/fatcat_web/routes.py277
-rw-r--r--python/fatcat_web/search.py28
12 files changed, 347 insertions, 297 deletions
diff --git a/python/fatcat_web/__init__.py b/python/fatcat_web/__init__.py
index 0d3445ca..5cb56d0b 100644
--- a/python/fatcat_web/__init__.py
+++ b/python/fatcat_web/__init__.py
@@ -55,7 +55,7 @@ app.jinja_env.trim_blocks = True
app.jinja_env.lstrip_blocks = True
-def auth_api(token):
+def auth_api(token: str) -> fatcat_openapi_client.DefaultApi:
conf = fatcat_openapi_client.Configuration()
conf.api_key["Authorization"] = token
conf.api_key_prefix["Authorization"] = "Bearer"
diff --git a/python/fatcat_web/auth.py b/python/fatcat_web/auth.py
index 4fe85770..2b3f5871 100644
--- a/python/fatcat_web/auth.py
+++ b/python/fatcat_web/auth.py
@@ -1,4 +1,5 @@
from collections import namedtuple
+from typing import Any, Dict, Optional
import fatcat_openapi_client
import pymacaroons
@@ -6,10 +7,10 @@ import requests
from flask import abort, flash, redirect, render_template, session
from flask_login import UserMixin, login_user, logout_user
-from fatcat_web import Config, api, app, login_manager, priv_api
+from fatcat_web import AnyResponse, Config, api, app, login_manager, priv_api
-def handle_logout():
+def handle_logout() -> None:
logout_user()
for k in ("editor", "api_token"):
if k in session:
@@ -17,7 +18,7 @@ def handle_logout():
session.clear()
-def handle_token_login(token):
+def handle_token_login(token: str) -> AnyResponse:
try:
m = pymacaroons.Macaroon.deserialize(token)
except pymacaroons.exceptions.MacaroonDeserializationException:
@@ -51,7 +52,7 @@ def handle_token_login(token):
# This will need to login/signup via fatcatd API, then set token in session
-def handle_oauth(remote, token, user_info):
+def handle_oauth(remote: Any, token: Optional[str], user_info: Dict[str, Any]) -> AnyResponse:
if user_info:
# fetch api login/signup using user_info
# ISS is basically the API url (though more formal in OIDC)
@@ -98,7 +99,7 @@ def handle_oauth(remote, token, user_info):
raise Exception("didn't receive OAuth user_info")
-def handle_ia_xauth(email, password):
+def handle_ia_xauth(email: str, password: str) -> AnyResponse:
resp = requests.post(
Config.IA_XAUTH_URI,
params={"op": "authenticate"},
@@ -153,7 +154,7 @@ def handle_ia_xauth(email, password):
return handle_oauth(remote, None, oauth_info)
-def handle_wmoauth(username):
+def handle_wmoauth(username: str) -> AnyResponse:
# pass off "as if" we did OAuth successfully
FakeOAuthRemote = namedtuple("FakeOAuthRemote", ["name", "OAUTH_CONFIG"])
remote = FakeOAuthRemote(
@@ -169,7 +170,7 @@ def handle_wmoauth(username):
@login_manager.user_loader
-def load_user(editor_id):
+def load_user(editor_id: str) -> UserMixin:
# looks for extra info in session, and updates the user object with that.
# If session isn't loaded/valid, should return None
if (not session.get("editor")) or (not session.get("api_token")):
diff --git a/python/fatcat_web/cors.py b/python/fatcat_web/cors.py
index bb32f7c2..b0f33760 100644
--- a/python/fatcat_web/cors.py
+++ b/python/fatcat_web/cors.py
@@ -5,18 +5,19 @@ This snippet from: http://flask.pocoo.org/snippets/56/
from datetime import timedelta
from functools import update_wrapper
+from typing import Any
from flask import current_app, make_response, request
def crossdomain(
- origin=None,
- methods=None,
- headers=None,
- max_age=21600,
- attach_to_all=True,
- automatic_options=True,
-):
+ origin: Any = None,
+ methods: Any = None,
+ headers: Any = None,
+ max_age: Any = 21600,
+ attach_to_all: bool = True,
+ automatic_options: bool = True,
+) -> Any:
if methods is not None:
methods = ", ".join(sorted(x.upper() for x in methods))
if headers is not None and not isinstance(headers, str):
@@ -26,15 +27,15 @@ def crossdomain(
if isinstance(max_age, timedelta):
max_age = max_age.total_seconds()
- def get_methods():
+ def get_methods() -> Any:
if methods is not None:
return methods
options_resp = current_app.make_default_options_response()
return options_resp.headers["allow"]
- def decorator(f):
- def wrapped_function(*args, **kwargs):
+ def decorator(f: Any) -> Any:
+ def wrapped_function(*args, **kwargs) -> Any:
if automatic_options and request.method == "OPTIONS":
resp = current_app.make_default_options_response()
else:
diff --git a/python/fatcat_web/editing_routes.py b/python/fatcat_web/editing_routes.py
index 03668e1e..32bff51d 100644
--- a/python/fatcat_web/editing_routes.py
+++ b/python/fatcat_web/editing_routes.py
@@ -1,6 +1,7 @@
-from typing import Optional
+from typing import Any, Optional
from fatcat_openapi_client import (
+ ApiClient,
ContainerEntity,
CreatorEntity,
Editgroup,
@@ -16,7 +17,7 @@ from flask import abort, flash, redirect, render_template, session
from flask_login import login_required
from fatcat_tools.transforms import entity_from_toml
-from fatcat_web import api, app, auth_api
+from fatcat_web import AnyResponse, api, app, auth_api
from fatcat_web.entity_helpers import generic_get_editgroup_entity, generic_get_entity
from fatcat_web.forms import (
ContainerEntityForm,
@@ -30,7 +31,7 @@ from fatcat_web.forms import (
def generic_entity_create_from_toml(
- user_api, entity_type: str, editgroup_id: str, toml_str: str
+ user_api: ApiClient, entity_type: str, editgroup_id: str, toml_str: str
) -> EntityEdit:
if entity_type == "container":
entity = entity_from_toml(toml_str, ContainerEntity)
@@ -59,7 +60,7 @@ def generic_entity_create_from_toml(
def generic_entity_delete_edit(
- user_api, entity_type: str, editgroup_id: str, edit_id: str
+ user_api: ApiClient, entity_type: str, editgroup_id: str, edit_id: str
) -> None:
try:
if entity_type == "container":
@@ -86,7 +87,7 @@ def generic_entity_delete_edit(
def generic_entity_delete_entity(
- user_api, entity_type: str, editgroup_id: str, entity_ident: str
+ user_api: ApiClient, entity_type: str, editgroup_id: str, entity_ident: str
) -> EntityEdit:
try:
if entity_type == "container":
@@ -111,7 +112,7 @@ def generic_entity_delete_entity(
def generic_entity_update_from_toml(
- user_api, entity_type: str, editgroup_id: str, existing_ident, toml_str: str
+ user_api: ApiClient, entity_type: str, editgroup_id: str, existing_ident: str, toml_str: str
) -> EntityEdit:
if entity_type == "container":
entity = entity_from_toml(toml_str, ContainerEntity)
@@ -139,7 +140,7 @@ def generic_entity_update_from_toml(
return edit
-def form_editgroup_get_or_create(api, edit_form):
+def form_editgroup_get_or_create(api: ApiClient, edit_form: Any) -> Optional[Editgroup]:
"""
This function expects a submitted, validated edit form
"""
@@ -168,7 +169,12 @@ def form_editgroup_get_or_create(api, edit_form):
return eg
-def generic_entity_edit(editgroup_id, entity_type, existing_ident, edit_template):
+def generic_entity_edit(
+ editgroup_id: Optional[str],
+ entity_type: str,
+ existing_ident: Optional[str],
+ edit_template: str,
+) -> AnyResponse:
"""
existing (entity)
@@ -342,7 +348,12 @@ def generic_entity_edit(editgroup_id, entity_type, existing_ident, edit_template
)
-def generic_entity_toml_edit(editgroup_id, entity_type, existing_ident, edit_template):
+def generic_entity_toml_edit(
+ editgroup_id: Optional[str],
+ entity_type: Any,
+ existing_ident: Optional[str],
+ edit_template: str,
+) -> AnyResponse:
"""
Similar to generic_entity_edit(), but for TOML editing mode.
@@ -469,7 +480,9 @@ def generic_entity_toml_edit(editgroup_id, entity_type, existing_ident, edit_tem
)
-def generic_entity_delete(editgroup_id: Optional[str], entity_type: str, existing_ident: str):
+def generic_entity_delete(
+ editgroup_id: Optional[str], entity_type: str, existing_ident: str
+) -> AnyResponse:
"""
Similar to generic_entity_edit(), but for deleting entities. This is a bit
simpler!
@@ -570,7 +583,9 @@ def generic_entity_delete(editgroup_id: Optional[str], entity_type: str, existin
)
-def generic_edit_delete(editgroup_id, entity_type, edit_id):
+def generic_edit_delete(
+ editgroup_id: Optional[str], entity_type: Any, edit_id: str
+) -> AnyResponse:
# fetch editgroup (if set) or 404
editgroup = None
if editgroup_id:
@@ -597,177 +612,177 @@ def generic_edit_delete(editgroup_id, entity_type, edit_id):
@app.route("/container/create", methods=["GET", "POST"])
@login_required
-def container_create_view():
+def container_create_view() -> AnyResponse:
return generic_entity_edit(None, "container", None, "container_create.html")
@app.route("/container/<ident>/edit", methods=["GET", "POST"])
@login_required
-def container_edit_view(ident):
+def container_edit_view(ident: str) -> AnyResponse:
return generic_entity_edit(None, "container", ident, "container_edit.html")
@app.route("/container/<ident>/delete", methods=["GET", "POST"])
@login_required
-def container_delete_view(ident):
+def container_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "container", ident)
@app.route("/editgroup/<editgroup_id>/container/<ident>/edit", methods=["GET", "POST"])
@login_required
-def container_editgroup_edit_view(editgroup_id, ident):
+def container_editgroup_edit_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_edit(editgroup_id, "container", ident, "container_edit.html")
@app.route("/editgroup/<editgroup_id>/container/<ident>/delete", methods=["GET", "POST"])
@login_required
-def container_editgroup_delete_view(editgroup_id, ident):
+def container_editgroup_delete_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "container", ident)
@app.route("/editgroup/<editgroup_id>/container/edit/<edit_id>/delete", methods=["POST"])
@login_required
-def container_edit_delete(editgroup_id, edit_id):
+def container_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "container", edit_id)
@app.route("/creator/<ident>/delete", methods=["GET", "POST"])
@login_required
-def creator_delete_view(ident):
+def creator_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "creator", ident)
@app.route("/editgroup/<editgroup_id>/creator/edit/<edit_id>/delete", methods=["POST"])
-def creator_edit_delete(editgroup_id, edit_id):
+def creator_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "creator", edit_id)
@app.route("/editgroup/<editgroup_id>/creator/<ident>/delete", methods=["GET", "POST"])
@login_required
-def creator_editgroup_delete(editgroup_id, ident):
+def creator_editgroup_delete(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "creator", ident)
@app.route("/file/create", methods=["GET", "POST"])
@login_required
-def file_create_view():
+def file_create_view() -> AnyResponse:
return generic_entity_edit(None, "file", None, "file_create.html")
@app.route("/file/<ident>/edit", methods=["GET", "POST"])
@login_required
-def file_edit_view(ident):
+def file_edit_view(ident: str) -> AnyResponse:
return generic_entity_edit(None, "file", ident, "file_edit.html")
@app.route("/file/<ident>/delete", methods=["GET", "POST"])
@login_required
-def file_delete_view(ident):
+def file_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "file", ident)
@app.route("/editgroup/<editgroup_id>/file/<ident>/edit", methods=["GET", "POST"])
@login_required
-def file_editgroup_edit_view(editgroup_id, ident):
+def file_editgroup_edit_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_edit(editgroup_id, "file", ident, "file_edit.html")
@app.route("/editgroup/<editgroup_id>/file/<ident>/delete", methods=["GET", "POST"])
@login_required
-def file_editgroup_delete_view(editgroup_id, ident):
+def file_editgroup_delete_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "file", ident)
@app.route("/editgroup/<editgroup_id>/file/edit/<edit_id>/delete", methods=["POST"])
@login_required
-def file_edit_delete(editgroup_id, edit_id):
+def file_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "file", edit_id)
@app.route("/fileset/<ident>/delete", methods=["GET", "POST"])
@login_required
-def fileset_delete_view(ident):
+def fileset_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "fileset", ident)
@app.route("/editgroup/<editgroup_id>/fileset/edit/<edit_id>/delete", methods=["POST"])
-def fileset_edit_delete(editgroup_id, edit_id):
+def fileset_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "fileset", edit_id)
@app.route("/editgroup/<editgroup_id>/fileset/<ident>/delete", methods=["GET", "POST"])
@login_required
-def fileset_editgroup_delete(editgroup_id, ident):
+def fileset_editgroup_delete(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "fileset", ident)
@app.route("/webcapture/<ident>/delete", methods=["GET", "POST"])
@login_required
-def webcapture_delete_view(ident):
+def webcapture_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "webcapture", ident)
@app.route("/editgroup/<editgroup_id>/webcapture/edit/<edit_id>/delete", methods=["POST"])
-def webcapture_edit_delete(editgroup_id, edit_id):
+def webcapture_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "webcapture", edit_id)
@app.route("/editgroup/<editgroup_id>/webcapture/<ident>/delete", methods=["GET", "POST"])
@login_required
-def webcapture_editgroup_delete(editgroup_id, ident):
+def webcapture_editgroup_delete(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "webcapture", ident)
@app.route("/release/create", methods=["GET", "POST"])
@login_required
-def release_create_view():
+def release_create_view() -> AnyResponse:
return generic_entity_edit(None, "release", None, "release_create.html")
@app.route("/release/<ident>/edit", methods=["GET", "POST"])
@login_required
-def release_edit_view(ident):
+def release_edit_view(ident: str) -> AnyResponse:
return generic_entity_edit(None, "release", ident, "release_edit.html")
@app.route("/release/<ident>/delete", methods=["GET", "POST"])
@login_required
-def release_delete_view(ident):
+def release_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "release", ident)
@app.route("/editgroup/<editgroup_id>/release/<ident>/edit", methods=["GET", "POST"])
@login_required
-def release_editgroup_edit(editgroup_id, ident):
+def release_editgroup_edit(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_edit(editgroup_id, "release", ident, "release_edit.html")
@app.route("/editgroup/<editgroup_id>/release/<ident>/delete", methods=["GET", "POST"])
@login_required
-def release_editgroup_delete(editgroup_id, ident):
+def release_editgroup_delete(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "release", ident)
@app.route("/editgroup/<editgroup_id>/release/edit/<edit_id>/delete", methods=["POST"])
@login_required
-def release_edit_delete(editgroup_id, edit_id):
+def release_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "release", edit_id)
@app.route("/work/<ident>/delete", methods=["GET", "POST"])
@login_required
-def work_delete_view(ident):
+def work_delete_view(ident: str) -> AnyResponse:
return generic_entity_delete(None, "work", ident)
@app.route("/editgroup/<editgroup_id>/work/edit/<edit_id>/delete", methods=["POST"])
-def work_edit_delete(editgroup_id, edit_id):
+def work_edit_delete(editgroup_id: str, edit_id: str) -> AnyResponse:
return generic_edit_delete(editgroup_id, "work", edit_id)
@app.route("/editgroup/<editgroup_id>/work/<ident>/delete", methods=["GET", "POST"])
@login_required
-def work_editgroup_delete(editgroup_id, ident):
+def work_editgroup_delete(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_delete(editgroup_id, "work", ident)
@@ -776,127 +791,127 @@ def work_editgroup_delete(editgroup_id, ident):
@app.route("/container/create/toml", methods=["GET", "POST"])
@login_required
-def container_create_toml_view():
+def container_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "container", None, "entity_create_toml.html")
@app.route("/container/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def container_edit_toml_view(ident):
+def container_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "container", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/container/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def container_editgroup_edit_toml(editgroup_id, ident):
+def container_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "container", ident, "entity_edit_toml.html")
@app.route("/creator/create/toml", methods=["GET", "POST"])
@login_required
-def creator_create_toml_view():
+def creator_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "creator", None, "entity_create_toml.html")
@app.route("/creator/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def creator_edit_toml_view(ident):
+def creator_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "creator", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/creator/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def creator_editgroup_edit_toml(editgroup_id, ident):
+def creator_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "creator", ident, "entity_edit_toml.html")
@app.route("/file/create/toml", methods=["GET", "POST"])
@login_required
-def file_create_toml_view():
+def file_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "file", None, "entity_create_toml.html")
@app.route("/file/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def file_edit_toml_view(ident):
+def file_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "file", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/file/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def file_editgroup_edit_toml(editgroup_id, ident):
+def file_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "file", ident, "entity_edit_toml.html")
@app.route("/fileset/create/toml", methods=["GET", "POST"])
@login_required
-def fileset_create_toml_view():
+def fileset_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "fileset", None, "entity_create_toml.html")
@app.route("/fileset/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def fileset_edit_toml_view(ident):
+def fileset_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "fileset", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/fileset/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def fileset_editgroup_edit_toml(editgroup_id, ident):
+def fileset_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "fileset", ident, "entity_edit_toml.html")
@app.route("/webcapture/create/toml", methods=["GET", "POST"])
@login_required
-def webcapture_create_toml_view():
+def webcapture_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "webcapture", None, "entity_create_toml.html")
@app.route("/webcapture/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def webcapture_edit_toml_view(ident):
+def webcapture_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "webcapture", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/webcapture/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def webcapture_editgroup_edit_toml(editgroup_id, ident):
+def webcapture_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "webcapture", ident, "entity_edit_toml.html")
@app.route("/release/create/toml", methods=["GET", "POST"])
@login_required
-def release_create_toml_view():
+def release_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "release", None, "entity_create_toml.html")
@app.route("/release/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def release_edit_toml_view(ident):
+def release_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "release", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/release/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def release_editgroup_edit_toml(editgroup_id, ident):
+def release_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "release", ident, "entity_edit_toml.html")
@app.route("/work/create/toml", methods=["GET", "POST"])
@login_required
-def work_create_toml_view():
+def work_create_toml_view() -> AnyResponse:
return generic_entity_toml_edit(None, "work", None, "entity_create_toml.html")
@app.route("/work/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def work_edit_toml_view(ident):
+def work_edit_toml_view(ident: str) -> AnyResponse:
return generic_entity_toml_edit(None, "work", ident, "entity_edit_toml.html")
@app.route("/editgroup/<editgroup_id>/work/<ident>/edit/toml", methods=["GET", "POST"])
@login_required
-def work_editgroup_edit_toml(editgroup_id, ident):
+def work_editgroup_edit_toml(editgroup_id: str, ident: str) -> AnyResponse:
return generic_entity_toml_edit(editgroup_id, "work", ident, "entity_edit_toml.html")
@@ -905,71 +920,71 @@ def work_editgroup_edit_toml(editgroup_id, ident):
@app.route("/creator/create", methods=["GET"])
@login_required
-def creator_create_view():
+def creator_create_view() -> AnyResponse:
return redirect("/creator/create/toml")
@app.route("/creator/<ident>/edit", methods=["GET"])
@login_required
-def creator_edit_view(ident):
+def creator_edit_view(ident: str) -> AnyResponse:
return redirect(f"/creator/{ident}/edit/toml")
@app.route("/editgroup/<editgroup_id>/creator/<ident>/edit", methods=["GET", "POST"])
@login_required
-def creator_editgroup_edit(editgroup_id, ident):
+def creator_editgroup_edit(editgroup_id: str, ident: str) -> AnyResponse:
return redirect(f"/editgroup/{editgroup_id}/creator/{ident}/edit/toml")
@app.route("/fileset/create", methods=["GET"])
@login_required
-def fileset_create_view():
+def fileset_create_view() -> AnyResponse:
return redirect("/fileset/create/toml")
@app.route("/fileset/<ident>/edit", methods=["GET"])
@login_required
-def fileset_edit_view(ident):
+def fileset_edit_view(ident: str) -> AnyResponse:
return redirect(f"/fileset/{ident}/edit/toml")
@app.route("/editgroup/<editgroup_id>/fileset/<ident>/edit", methods=["GET", "POST"])
@login_required
-def fileset_editgroup_edit(editgroup_id, ident):
+def fileset_editgroup_edit(editgroup_id: str, ident: str) -> AnyResponse:
return redirect(f"/editgroup/{editgroup_id}/fileset/{ident}/edit/toml")
@app.route("/webcapture/create", methods=["GET"])
@login_required
-def webcapture_create_view():
+def webcapture_create_view() -> AnyResponse:
return redirect("/webcapture/create/toml")
@app.route("/webcapture/<ident>/edit", methods=["GET"])
@login_required
-def webcapture_edit_view(ident):
+def webcapture_edit_view(ident: str) -> AnyResponse:
return redirect(f"/webcapture/{ident}/edit/toml")
@app.route("/editgroup/<editgroup_id>/webcapture/<ident>/edit", methods=["GET", "POST"])
@login_required
-def webcapture_editgroup_edit(editgroup_id, ident):
+def webcapture_editgroup_edit(editgroup_id: str, ident: str) -> AnyResponse:
return redirect(f"/editgroup/{editgroup_id}/webcapture/{ident}/edit/toml")
@app.route("/work/create", methods=["GET"])
@login_required
-def work_create_view():
+def work_create_view() -> AnyResponse:
return redirect("/work/create/toml")
@app.route("/work/<ident>/edit", methods=["GET"])
@login_required
-def work_edit_view(ident):
+def work_edit_view(ident: str) -> AnyResponse:
return redirect(f"/work/{ident}/edit/toml")
@app.route("/editgroup/<editgroup_id>/work/<ident>/edit", methods=["GET", "POST"])
@login_required
-def work_editgroup_edit(editgroup_id, ident):
+def work_editgroup_edit(editgroup_id: str, ident: str) -> AnyResponse:
return redirect(f"/editgroup/{editgroup_id}/work/{ident}/edit/toml")
diff --git a/python/fatcat_web/entity_helpers.py b/python/fatcat_web/entity_helpers.py
index dbe11cb4..86543ee3 100644
--- a/python/fatcat_web/entity_helpers.py
+++ b/python/fatcat_web/entity_helpers.py
@@ -1,6 +1,10 @@
+from typing import Any, Tuple
+
from fatcat_openapi_client import (
ContainerEntity,
CreatorEntity,
+ Editgroup,
+ EntityEdit,
FileEntity,
FilesetEntity,
ReleaseEntity,
@@ -20,7 +24,7 @@ from fatcat_web import api
from fatcat_web.hacks import strip_extlink_xml, wayback_suffix
-def enrich_container_entity(entity):
+def enrich_container_entity(entity: ContainerEntity) -> ContainerEntity:
if entity.state in ("redirect", "deleted"):
return entity
if entity.state == "active":
@@ -28,7 +32,7 @@ def enrich_container_entity(entity):
return entity
-def enrich_creator_entity(entity):
+def enrich_creator_entity(entity: CreatorEntity) -> CreatorEntity:
if entity.state in ("redirect", "deleted"):
return entity
entity._releases = None
@@ -37,13 +41,13 @@ def enrich_creator_entity(entity):
return entity
-def enrich_file_entity(entity):
+def enrich_file_entity(entity: FileEntity) -> FileEntity:
if entity.state == "active":
entity._es = file_to_elasticsearch(entity)
return entity
-def enrich_fileset_entity(entity):
+def enrich_fileset_entity(entity: FilesetEntity) -> FilesetEntity:
if entity.state in ("redirect", "deleted"):
return entity
entity._total_size = None
@@ -52,14 +56,14 @@ def enrich_fileset_entity(entity):
return entity
-def enrich_webcapture_entity(entity):
+def enrich_webcapture_entity(entity: WebcaptureEntity) -> WebcaptureEntity:
if entity.state in ("redirect", "deleted"):
return entity
entity._wayback_suffix = wayback_suffix(entity)
return entity
-def enrich_release_entity(entity):
+def enrich_release_entity(entity: ReleaseEntity) -> ReleaseEntity:
if entity.state in ("redirect", "deleted"):
return entity
if entity.state == "active":
@@ -126,7 +130,7 @@ def enrich_release_entity(entity):
return entity
-def enrich_work_entity(entity):
+def enrich_work_entity(entity: WorkEntity) -> WorkEntity:
if entity.state in ("redirect", "deleted"):
return entity
entity._releases = None
@@ -135,7 +139,7 @@ def enrich_work_entity(entity):
return entity
-def generic_get_entity(entity_type, ident):
+def generic_get_entity(entity_type: str, ident: str) -> Any:
try:
if entity_type == "container":
return enrich_container_entity(api.get_container(ident))
@@ -161,7 +165,7 @@ def generic_get_entity(entity_type, ident):
abort(400)
-def generic_get_entity_revision(entity_type, revision_id):
+def generic_get_entity_revision(entity_type: str, revision_id: str) -> Any:
try:
if entity_type == "container":
return enrich_container_entity(api.get_container_revision(revision_id))
@@ -191,9 +195,9 @@ def generic_get_entity_revision(entity_type, revision_id):
abort(400)
-def generic_deleted_entity(entity_type, ident):
+def generic_deleted_entity(entity_type: str, ident: str) -> Any:
if entity_type == "container":
- entity = ContainerEntity()
+ entity: Any = ContainerEntity()
elif entity_type == "creator":
entity = CreatorEntity()
elif entity_type == "file":
@@ -212,7 +216,9 @@ def generic_deleted_entity(entity_type, ident):
return entity
-def generic_get_editgroup_entity(editgroup, entity_type, ident):
+def generic_get_editgroup_entity(
+ editgroup: Editgroup, entity_type: str, ident: str
+) -> Tuple[Any, EntityEdit]:
if entity_type == "container":
edits = editgroup.edits.containers
elif entity_type == "creator":
diff --git a/python/fatcat_web/forms.py b/python/fatcat_web/forms.py
index 25bfbb90..9f11c423 100644
--- a/python/fatcat_web/forms.py
+++ b/python/fatcat_web/forms.py
@@ -4,6 +4,7 @@ but can't find one that is actually maintained.
"""
import datetime
+from typing import Any, Dict, List, Tuple
import toml
from fatcat_openapi_client import (
@@ -30,7 +31,7 @@ from wtforms import (
from fatcat_tools.transforms import entity_to_toml
-release_type_options = [
+release_type_options: List[Tuple[str, str]] = [
("", "Unknown (blank)"),
("article-journal", "Journal Article"),
("paper-conference", "Conference Proceeding"),
@@ -40,7 +41,7 @@ release_type_options = [
("dataset", "Dataset"),
("stub", "Invalid/Stub"),
]
-release_stage_options = [
+release_stage_options: List[Tuple[str, str]] = [
("", "Unknown (blank)"),
("draft", "Draft"),
("submitted", "Submitted"),
@@ -48,7 +49,7 @@ release_stage_options = [
("published", "Published"),
("updated", "Updated"),
]
-withdrawn_status_options = [
+withdrawn_status_options: List[Tuple[str, str]] = [
("", "Not Withdrawn (blank)"),
("retracted", "Retracted"),
("withdrawn", "Withdrawn"),
@@ -58,7 +59,7 @@ withdrawn_status_options = [
("safety", "Public Safety"),
("national-security", "National Security"),
]
-role_type_options = [
+role_type_options: List[Tuple[str, str]] = [
("author", "Author"),
("editor", "Editor"),
("translator", "Translator"),
@@ -87,7 +88,7 @@ class ReleaseContribForm(FlaskForm):
role = SelectField([validators.DataRequired()], choices=role_type_options, default="author")
-RELEASE_SIMPLE_ATTRS = [
+RELEASE_SIMPLE_ATTRS: List[str] = [
"title",
"original_title",
"work_id",
@@ -105,17 +106,17 @@ RELEASE_SIMPLE_ATTRS = [
"license_slug",
]
-RELEASE_EXTID_ATTRS = ["doi", "wikidata_qid", "isbn13", "pmid", "pmcid"]
+RELEASE_EXTID_ATTRS: List[str] = ["doi", "wikidata_qid", "isbn13", "pmid", "pmcid"]
-def valid_year(form, field):
+def valid_year(form: Any, field: Any) -> None:
if field.data > datetime.date.today().year + 5:
raise ValidationError(f"Year is too far in the future: {field.data}")
if field.data < 10:
raise ValidationError(f"Year is too far in the past: {field.data}")
-def valid_2char_ascii(form, field):
+def valid_2char_ascii(form: Any, field: Any) -> None:
if (
len(field.data) != 2
or len(field.data.encode("utf-8")) != 2
@@ -179,7 +180,7 @@ class ReleaseEntityForm(EntityEditForm):
# abstracts
@staticmethod
- def from_entity(re):
+ def from_entity(re: ReleaseEntity) -> "ReleaseEntityForm":
"""
Initializes form with values from an existing release entity.
"""
@@ -198,13 +199,13 @@ class ReleaseEntityForm(EntityEditForm):
ref.contribs.append_entry(rcf)
return ref
- def to_entity(self):
+ def to_entity(self) -> ReleaseEntity:
assert self.title.data
entity = ReleaseEntity(title=self.title.data, ext_ids=ReleaseExtIds())
self.update_entity(entity)
return entity
- def update_entity(self, re):
+ def update_entity(self, re: ReleaseEntity) -> None:
"""
Mutates a release entity in place, updating fields with values from
this form.
@@ -249,7 +250,7 @@ class ReleaseEntityForm(EntityEditForm):
re.edit_extra = dict(description=self.edit_description.data)
-container_type_options = (
+container_type_options: List[Tuple[str, str]] = [
("", "Unknown (blank)"),
("journal", "Scholarly Journal"),
("proceedings", "Proceedings"),
@@ -258,9 +259,9 @@ container_type_options = (
("magazine", "Magazine"),
("trade", "Trade Magazine"),
("test", "Test / Dummy"),
-)
+]
-CONTAINER_SIMPLE_ATTRS = [
+CONTAINER_SIMPLE_ATTRS: List[str] = [
"name",
"container_type",
"publisher",
@@ -269,7 +270,7 @@ CONTAINER_SIMPLE_ATTRS = [
"issne",
"issnp",
]
-CONTAINER_EXTRA_ATTRS = ["original_name", "country"]
+CONTAINER_EXTRA_ATTRS: List[str] = ["original_name", "country"]
class ContainerEntityForm(EntityEditForm):
@@ -296,7 +297,7 @@ class ContainerEntityForm(EntityEditForm):
)
@staticmethod
- def from_entity(ce):
+ def from_entity(ce: ContainerEntity) -> "ContainerEntityForm":
"""
Initializes form with values from an existing container entity.
"""
@@ -314,13 +315,13 @@ class ContainerEntityForm(EntityEditForm):
cef.urls.append_entry(url)
return cef
- def to_entity(self):
+ def to_entity(self) -> ContainerEntity:
assert self.name.data
entity = ContainerEntity(name=self.name.data)
self.update_entity(entity)
return entity
- def update_entity(self, ce):
+ def update_entity(self, ce: ContainerEntity) -> None:
"""
Mutates a container entity in place, updating fields with values from
this form.
@@ -350,7 +351,7 @@ class ContainerEntityForm(EntityEditForm):
ce.extra = None
-url_rel_options = [
+url_rel_options: List[Tuple[str, str]] = [
("web", "Public Web"),
("webarchive", "Web Archive"),
("repository", "Repository"),
@@ -361,7 +362,7 @@ url_rel_options = [
("aggregator", "Aggregator"),
]
-FILE_SIMPLE_ATTRS = ["size", "md5", "sha1", "sha256", "mimetype"]
+FILE_SIMPLE_ATTRS: List[str] = ["size", "md5", "sha1", "sha256", "mimetype"]
class FileUrlForm(FlaskForm):
@@ -392,7 +393,7 @@ class FileEntityForm(EntityEditForm):
)
@staticmethod
- def from_entity(fe):
+ def from_entity(fe: FileEntity) -> "FileEntityForm":
"""
Initializes form with values from an existing file entity.
"""
@@ -409,13 +410,13 @@ class FileEntityForm(EntityEditForm):
ref.release_ids.append_entry(r)
return ref
- def to_entity(self):
+ def to_entity(self) -> FileEntity:
assert self.sha1.data
entity = FileEntity()
self.update_entity(entity)
return entity
- def update_entity(self, fe):
+ def update_entity(self, fe: FileEntity) -> None:
"""
Mutates in place, updating fields with values from this form.
@@ -445,7 +446,7 @@ class FileEntityForm(EntityEditForm):
fe.edit_extra = dict(description=self.edit_description.data)
-INGEST_TYPE_OPTIONS = [
+INGEST_TYPE_OPTIONS: List[Tuple[str, str]] = [
("pdf", "PDF Fulltext"),
("html", "HTML Fulltext"),
("xml", "XML Fulltext"),
@@ -465,7 +466,9 @@ class SavePaperNowForm(FlaskForm):
default="",
)
- def to_ingest_request(self, release, ingest_request_source="savepapernow"):
+ def to_ingest_request(
+ self, release: ReleaseEntity, ingest_request_source: str = "savepapernow"
+ ) -> Dict[str, Any]:
base_url = self.base_url.data
ext_ids = release.ext_ids.to_dict()
# by default this dict has a bunch of empty values
@@ -496,7 +499,7 @@ class SavePaperNowForm(FlaskForm):
return ingest_request
-def valid_toml(form, field):
+def valid_toml(form: Any, field: Any) -> None:
try:
toml.loads(field.data)
except toml.TomlDecodeError as tpe:
@@ -514,7 +517,7 @@ class EntityTomlForm(EntityEditForm):
)
@staticmethod
- def from_entity(entity):
+ def from_entity(entity: Any) -> "EntityTomlForm":
"""
Initializes form with TOML version of existing entity
"""
@@ -569,7 +572,9 @@ class ReferenceMatchForm(FlaskForm):
release_stage = StringField("Release Stage")
@staticmethod
- def from_grobid_parse(parse_dict, raw_citation):
+ def from_grobid_parse(
+ parse_dict: Dict[str, Any], raw_citation: str
+ ) -> "ReferenceMatchForm":
"""
Initializes form from GROBID extraction
"""
diff --git a/python/fatcat_web/graphics.py b/python/fatcat_web/graphics.py
index 82a0a577..6f9bcd2f 100644
--- a/python/fatcat_web/graphics.py
+++ b/python/fatcat_web/graphics.py
@@ -4,7 +4,7 @@ import pygal
from pygal.style import CleanStyle
-def ia_coverage_histogram(rows: List[Tuple]) -> pygal.Graph:
+def ia_coverage_histogram(rows: List[Tuple[int, bool, int]]) -> pygal.Graph:
"""
Note: this returns a raw pygal chart; it does not render it to SVG/PNG
diff --git a/python/fatcat_web/hacks.py b/python/fatcat_web/hacks.py
index 06350b41..0339a0d7 100644
--- a/python/fatcat_web/hacks.py
+++ b/python/fatcat_web/hacks.py
@@ -1,15 +1,17 @@
import re
+from fatcat_openapi_client import WebcaptureEntity
+
STRIP_EXTLINK_XML_RE = re.compile(r"<ext-link.*xlink:type=\"simple\">")
-def strip_extlink_xml(unstr):
+def strip_extlink_xml(unstr: str) -> str:
unstr = unstr.replace("</ext-link>", "")
unstr = STRIP_EXTLINK_XML_RE.sub("", unstr)
return unstr
-def test_strip_extlink_xml():
+def test_strip_extlink_xml() -> None:
assert strip_extlink_xml("asdf") == "asdf"
assert (
strip_extlink_xml(
@@ -19,7 +21,7 @@ def test_strip_extlink_xml():
)
-def wayback_suffix(entity):
+def wayback_suffix(entity: WebcaptureEntity) -> str:
"""
Takes a webcapture entity and returns a suffix to be appended to wayback URLs
"""
diff --git a/python/fatcat_web/kafka.py b/python/fatcat_web/kafka.py
index 36dafade..a05b97e0 100644
--- a/python/fatcat_web/kafka.py
+++ b/python/fatcat_web/kafka.py
@@ -1,9 +1,13 @@
+from typing import Any, Dict, Optional
+
import requests
from fatcat_web import Config
-def kafka_pixy_produce(topic, msg, key=None, sync=True, timeout=25):
+def kafka_pixy_produce(
+ topic: str, msg: str, key: Optional[bytes] = None, sync: bool = True, timeout: float = 25
+) -> None:
"""
Simple helper to public a message to the given Kafka topic, via the
configured kafka-pixy HTTP gateway
@@ -17,7 +21,7 @@ def kafka_pixy_produce(topic, msg, key=None, sync=True, timeout=25):
if not Config.KAFKA_PIXY_ENDPOINT:
raise Exception("Kafka produce error: kafka-pixy endpoint not configured")
- params = dict()
+ params: Dict[str, Any] = dict()
if key:
params["key"] = key
if sync:
diff --git a/python/fatcat_web/ref_routes.py b/python/fatcat_web/ref_routes.py
index b45edf78..0f8c814c 100644
--- a/python/fatcat_web/ref_routes.py
+++ b/python/fatcat_web/ref_routes.py
@@ -4,6 +4,7 @@ Flask endpoints for reference (citation) endpoints. Eg, listing references
"""
import json
+from typing import Optional
from flask import Response, jsonify, render_template, request
from fuzzycat.grobid_unstructured import (
@@ -23,14 +24,18 @@ from fatcat_tools.references import (
)
from fatcat_tools.transforms.access import release_access_options
from fatcat_tools.transforms.entities import entity_to_dict
-from fatcat_web import api, app
+from fatcat_web import AnyResponse, api, app
from fatcat_web.cors import crossdomain
from fatcat_web.entity_helpers import generic_get_entity
from fatcat_web.forms import ReferenceMatchForm
def _refs_web(
- direction, release_ident=None, work_ident=None, openlibrary_id=None, wikipedia_article=None
+ direction: str,
+ release_ident: Optional[str] = None,
+ work_ident: Optional[str] = None,
+ openlibrary_id: Optional[str] = None,
+ wikipedia_article: Optional[str] = None,
) -> RefHitsEnriched:
offset_arg = request.args.get("offset", "0")
offset: int = max(0, int(offset_arg)) if offset_arg.isnumeric() else 0
@@ -74,7 +79,7 @@ def _refs_web(
@app.route("/release/<string(length=26):ident>/refs-in", methods=["GET"])
-def release_view_refs_inbound(ident):
+def release_view_refs_inbound(ident: str) -> AnyResponse:
if request.accept_mimetypes.best == "application/json":
return release_view_refs_inbound_json(ident)
@@ -89,7 +94,7 @@ def release_view_refs_inbound(ident):
@app.route("/release/<string(length=26):ident>/refs-out", methods=["GET"])
-def release_view_refs_outbound(ident):
+def release_view_refs_outbound(ident: str) -> AnyResponse:
if request.accept_mimetypes.best == "application/json":
return release_view_refs_outbound_json(ident)
@@ -104,7 +109,7 @@ def release_view_refs_outbound(ident):
@app.route("/openlibrary/OL<int:id_num>W/refs-in", methods=["GET"])
-def openlibrary_view_refs_inbound(id_num):
+def openlibrary_view_refs_inbound(id_num: int) -> AnyResponse:
if request.accept_mimetypes.best == "application/json":
return openlibrary_view_refs_inbound_json(id_num)
@@ -124,7 +129,7 @@ def openlibrary_view_refs_inbound(id_num):
@app.route(
"/wikipedia/<string(length=2):wiki_lang>:<string:wiki_article>/refs-out", methods=["GET"]
)
-def wikipedia_view_refs_outbound(wiki_lang: str, wiki_article: str):
+def wikipedia_view_refs_outbound(wiki_lang: str, wiki_article: str) -> AnyResponse:
if request.accept_mimetypes.best == "application/json":
return wikipedia_view_refs_outbound_json(wiki_lang, wiki_article)
@@ -146,7 +151,7 @@ def wikipedia_view_refs_outbound(wiki_lang: str, wiki_article: str):
@app.route("/reference/match", methods=["GET", "POST"])
-def reference_match():
+def reference_match() -> AnyResponse:
grobid_status = None
grobid_dict = None
@@ -230,21 +235,21 @@ def reference_match():
@app.route("/release/<string(length=26):ident>/refs-out.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def release_view_refs_outbound_json(ident):
+def release_view_refs_outbound_json(ident: str) -> AnyResponse:
hits = _refs_web("out", release_ident=ident)
return Response(hits.json(exclude_unset=True), mimetype="application/json")
@app.route("/release/<string(length=26):ident>/refs-in.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def release_view_refs_inbound_json(ident):
+def release_view_refs_inbound_json(ident: str) -> AnyResponse:
hits = _refs_web("in", release_ident=ident)
return Response(hits.json(exclude_unset=True), mimetype="application/json")
@app.route("/openlibrary/OL<int:id_num>W/refs-in.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def openlibrary_view_refs_inbound_json(id_num):
+def openlibrary_view_refs_inbound_json(id_num: int) -> AnyResponse:
openlibrary_id = f"OL{id_num}W"
hits = _refs_web("in", openlibrary_id=openlibrary_id)
return Response(hits.json(exclude_unset=True), mimetype="application/json")
@@ -255,7 +260,7 @@ def openlibrary_view_refs_inbound_json(id_num):
methods=["GET", "OPTIONS"],
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def wikipedia_view_refs_outbound_json(wiki_lang: str, wiki_article: str):
+def wikipedia_view_refs_outbound_json(wiki_lang: str, wiki_article: str) -> AnyResponse:
wiki_article = wiki_article.replace("_", " ")
wikipedia_article = wiki_lang + ":" + wiki_article
hits = _refs_web("out", wikipedia_article=wikipedia_article)
@@ -264,7 +269,7 @@ def wikipedia_view_refs_outbound_json(wiki_lang: str, wiki_article: str):
@app.route("/reference/match.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def reference_match_json():
+def reference_match_json() -> AnyResponse:
form = ReferenceMatchForm(request.args)
if form.validate():
if form.submit_type.data == "match":
diff --git a/python/fatcat_web/routes.py b/python/fatcat_web/routes.py
index e6963dbc..17921f30 100644
--- a/python/fatcat_web/routes.py
+++ b/python/fatcat_web/routes.py
@@ -1,5 +1,6 @@
import json
import os
+from typing import Any, Callable, Dict, List, Optional
import citeproc_styles
from fatcat_openapi_client import EditgroupAnnotation
@@ -30,7 +31,7 @@ from fatcat_tools.normal import (
clean_sha256,
)
from fatcat_tools.transforms import citeproc_csl, release_to_csl
-from fatcat_web import Config, api, app, auth_api, mwoauth, priv_api
+from fatcat_web import AnyResponse, Config, api, app, auth_api, mwoauth, priv_api
from fatcat_web.auth import (
handle_ia_xauth,
handle_logout,
@@ -73,7 +74,7 @@ from fatcat_web.search import (
@app.route("/container/<string(length=26):ident>/history", methods=["GET"])
-def container_history(ident):
+def container_history(ident: str) -> AnyResponse:
try:
entity = api.get_container(ident)
history = api.get_container_history(ident)
@@ -86,7 +87,7 @@ def container_history(ident):
@app.route("/creator/<string(length=26):ident>/history", methods=["GET"])
-def creator_history(ident):
+def creator_history(ident: str) -> AnyResponse:
try:
entity = api.get_creator(ident)
history = api.get_creator_history(ident)
@@ -98,7 +99,7 @@ def creator_history(ident):
@app.route("/file/<string(length=26):ident>/history", methods=["GET"])
-def file_history(ident):
+def file_history(ident: str) -> AnyResponse:
try:
entity = api.get_file(ident)
history = api.get_file_history(ident)
@@ -110,7 +111,7 @@ def file_history(ident):
@app.route("/fileset/<string(length=26):ident>/history", methods=["GET"])
-def fileset_history(ident):
+def fileset_history(ident: str) -> AnyResponse:
try:
entity = api.get_fileset(ident)
history = api.get_fileset_history(ident)
@@ -122,7 +123,7 @@ def fileset_history(ident):
@app.route("/webcapture/<string(length=26):ident>/history", methods=["GET"])
-def webcapture_history(ident):
+def webcapture_history(ident: str) -> AnyResponse:
try:
entity = api.get_webcapture(ident)
history = api.get_webcapture_history(ident)
@@ -134,7 +135,7 @@ def webcapture_history(ident):
@app.route("/release/<string(length=26):ident>/history", methods=["GET"])
-def release_history(ident):
+def release_history(ident: str) -> AnyResponse:
try:
entity = api.get_release(ident)
history = api.get_release_history(ident)
@@ -146,7 +147,7 @@ def release_history(ident):
@app.route("/work/<string(length=26):ident>/history", methods=["GET"])
-def work_history(ident):
+def work_history(ident: str) -> AnyResponse:
try:
entity = api.get_work(ident)
history = api.get_work_history(ident)
@@ -157,7 +158,9 @@ def work_history(ident):
)
-def generic_lookup_view(entity_type, lookup_template, extid_types, lookup_lambda):
+def generic_lookup_view(
+ entity_type: str, lookup_template: str, extid_types: List[str], lookup_lambda: Callable
+) -> AnyResponse:
extid = None
for key in extid_types:
if request.args.get(key):
@@ -195,48 +198,48 @@ def generic_lookup_view(entity_type, lookup_template, extid_types, lookup_lambda
@app.route("/container/lookup", methods=["GET"])
-def container_lookup():
+def container_lookup() -> AnyResponse:
return generic_lookup_view(
"container",
"container_lookup.html",
- ("issn", "issne", "issnp", "issnl", "wikidata_qid"),
+ ["issn", "issne", "issnp", "issnl", "wikidata_qid"],
lambda p: api.lookup_container(**p),
)
@app.route("/creator/lookup", methods=["GET"])
-def creator_lookup():
+def creator_lookup() -> AnyResponse:
return generic_lookup_view(
"creator",
"creator_lookup.html",
- ("orcid", "wikidata_qid"),
+ ["orcid", "wikidata_qid"],
lambda p: api.lookup_creator(**p),
)
@app.route("/file/lookup", methods=["GET"])
-def file_lookup():
+def file_lookup() -> AnyResponse:
return generic_lookup_view(
- "file", "file_lookup.html", ("md5", "sha1", "sha256"), lambda p: api.lookup_file(**p)
+ "file", "file_lookup.html", ["md5", "sha1", "sha256"], lambda p: api.lookup_file(**p)
)
@app.route("/fileset/lookup", methods=["GET"])
-def fileset_lookup():
+def fileset_lookup() -> AnyResponse:
abort(404)
@app.route("/webcapture/lookup", methods=["GET"])
-def webcapture_lookup():
+def webcapture_lookup() -> AnyResponse:
abort(404)
@app.route("/release/lookup", methods=["GET"])
-def release_lookup():
+def release_lookup() -> AnyResponse:
return generic_lookup_view(
"release",
"release_lookup.html",
- (
+ [
"doi",
"wikidata_qid",
"pmid",
@@ -249,20 +252,20 @@ def release_lookup():
"mag",
"oai",
"hdl",
- ),
+ ],
lambda p: api.lookup_release(**p),
)
@app.route("/work/lookup", methods=["GET"])
-def work_lookup():
+def work_lookup() -> AnyResponse:
abort(404)
### More Generic Entity Views ###############################################
-def generic_entity_view(entity_type, ident, view_template):
+def generic_entity_view(entity_type: str, ident: str, view_template: str) -> AnyResponse:
entity = generic_get_entity(entity_type, ident)
if entity.state == "redirect":
@@ -288,7 +291,9 @@ def generic_entity_view(entity_type, ident, view_template):
)
-def generic_entity_revision_view(entity_type, revision_id, view_template):
+def generic_entity_revision_view(
+ entity_type: str, revision_id: str, view_template: str
+) -> AnyResponse:
entity = generic_get_entity_revision(entity_type, revision_id)
metadata = entity.to_dict()
@@ -300,7 +305,9 @@ def generic_entity_revision_view(entity_type, revision_id, view_template):
)
-def generic_editgroup_entity_view(editgroup_id, entity_type, ident, view_template):
+def generic_editgroup_entity_view(
+ editgroup_id: Optional[str], entity_type: str, ident: str, view_template: str
+) -> AnyResponse:
try:
editgroup = api.get_editgroup(editgroup_id)
except ApiException as ae:
@@ -323,40 +330,40 @@ def generic_editgroup_entity_view(editgroup_id, entity_type, ident, view_templat
@app.route("/container/<string(length=26):ident>", methods=["GET"])
-def container_view(ident):
+def container_view(ident: str) -> AnyResponse:
return generic_entity_view("container", ident, "container_view.html")
@app.route("/container_<string(length=26):ident>", methods=["GET"])
-def container_underscore_view(ident):
+def container_underscore_view(ident: str) -> AnyResponse:
return redirect("/container/{}".format(ident))
@app.route("/container/<string(length=26):ident>/coverage", methods=["GET"])
-def container_view_coverage(ident):
+def container_view_coverage(ident: str) -> AnyResponse:
# note: there is a special hack to add entity._type_preservation for this endpoint
return generic_entity_view("container", ident, "container_view_coverage.html")
@app.route("/container/<string(length=26):ident>/metadata", methods=["GET"])
-def container_view_metadata(ident):
+def container_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("container", ident, "entity_view_metadata.html")
@app.route("/container/rev/<uuid:revision_id>", methods=["GET"])
-def container_revision_view(revision_id):
+def container_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("container", str(revision_id), "container_view.html")
@app.route("/container/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def container_revision_view_metadata(revision_id):
+def container_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"container", str(revision_id), "entity_view_metadata.html"
)
@app.route("/editgroup/<editgroup_id>/container/<string(length=26):ident>", methods=["GET"])
-def container_editgroup_view(editgroup_id, ident):
+def container_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "container", ident, "container_view.html"
)
@@ -365,160 +372,160 @@ def container_editgroup_view(editgroup_id, ident):
@app.route(
"/editgroup/<editgroup_id>/container/<string(length=26):ident>/metadata", methods=["GET"]
)
-def container_editgroup_view_metadata(editgroup_id, ident):
+def container_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "container", ident, "entity_view_metadata.html"
)
@app.route("/creator/<string(length=26):ident>", methods=["GET"])
-def creator_view(ident):
+def creator_view(ident: str) -> AnyResponse:
return generic_entity_view("creator", ident, "creator_view.html")
@app.route("/creator_<string(length=26):ident>", methods=["GET"])
-def creator_underscore_view(ident):
+def creator_underscore_view(ident: str) -> AnyResponse:
return redirect("/creator/{}".format(ident))
@app.route("/creator/<string(length=26):ident>/metadata", methods=["GET"])
-def creator_view_metadata(ident):
+def creator_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("creator", ident, "entity_view_metadata.html")
@app.route("/creator/rev/<uuid:revision_id>", methods=["GET"])
-def creator_revision_view(revision_id):
+def creator_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("creator", str(revision_id), "creator_view.html")
@app.route("/creator/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def creator_revision_view_metadata(revision_id):
+def creator_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"creator", str(revision_id), "entity_view_metadata.html"
)
@app.route("/editgroup/<editgroup_id>/creator/<string(length=26):ident>", methods=["GET"])
-def creator_editgroup_view(editgroup_id, ident):
+def creator_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(editgroup_id, "creator", ident, "creator_view.html")
@app.route(
"/editgroup/<editgroup_id>/creator/<string(length=26):ident>/metadata", methods=["GET"]
)
-def creator_editgroup_view_metadata(editgroup_id, ident):
+def creator_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "creator", ident, "entity_view_metadata.html"
)
@app.route("/file/<string(length=26):ident>", methods=["GET"])
-def file_view(ident):
+def file_view(ident: str) -> AnyResponse:
return generic_entity_view("file", ident, "file_view.html")
@app.route("/file_<string(length=26):ident>", methods=["GET"])
-def file_underscore_view(ident):
+def file_underscore_view(ident: str) -> AnyResponse:
return redirect("/file/{}".format(ident))
@app.route("/file/<string(length=26):ident>/metadata", methods=["GET"])
-def file_view_metadata(ident):
+def file_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("file", ident, "entity_view_metadata.html")
@app.route("/file/rev/<uuid:revision_id>", methods=["GET"])
-def file_revision_view(revision_id):
+def file_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("file", str(revision_id), "file_view.html")
@app.route("/file/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def file_revision_view_metadata(revision_id):
+def file_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("file", str(revision_id), "entity_view_metadata.html")
@app.route("/editgroup/<editgroup_id>/file/<string(length=26):ident>", methods=["GET"])
-def file_editgroup_view(editgroup_id, ident):
+def file_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(editgroup_id, "file", ident, "file_view.html")
@app.route("/editgroup/<editgroup_id>/file/<string(length=26):ident>/metadata", methods=["GET"])
-def file_editgroup_view_metadata(editgroup_id, ident):
+def file_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "file", ident, "entity_view_metadata.html"
)
@app.route("/fileset/<string(length=26):ident>", methods=["GET"])
-def fileset_view(ident):
+def fileset_view(ident: str) -> AnyResponse:
return generic_entity_view("fileset", ident, "fileset_view.html")
@app.route("/fileset_<string(length=26):ident>", methods=["GET"])
-def fileset_underscore_view(ident):
+def fileset_underscore_view(ident: str) -> AnyResponse:
return redirect("/fileset/{}".format(ident))
@app.route("/fileset/<string(length=26):ident>/metadata", methods=["GET"])
-def fileset_view_metadata(ident):
+def fileset_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("fileset", ident, "entity_view_metadata.html")
@app.route("/fileset/rev/<uuid:revision_id>", methods=["GET"])
-def fileset_revision_view(revision_id):
+def fileset_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("fileset", str(revision_id), "fileset_view.html")
@app.route("/fileset/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def fileset_revision_view_metadata(revision_id):
+def fileset_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"fileset", str(revision_id), "entity_view_metadata.html"
)
@app.route("/editgroup/<editgroup_id>/fileset/<string(length=26):ident>", methods=["GET"])
-def fileset_editgroup_view(editgroup_id, ident):
+def fileset_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(editgroup_id, "fileset", ident, "fileset_view.html")
@app.route(
"/editgroup/<editgroup_id>/fileset/<string(length=26):ident>/metadata", methods=["GET"]
)
-def fileset_editgroup_view_metadata(editgroup_id, ident):
+def fileset_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "fileset", ident, "entity_view_metadata.html"
)
@app.route("/webcapture/<string(length=26):ident>", methods=["GET"])
-def webcapture_view(ident):
+def webcapture_view(ident: str) -> AnyResponse:
return generic_entity_view("webcapture", ident, "webcapture_view.html")
@app.route("/webcapture_<string(length=26):ident>", methods=["GET"])
-def webcapture_underscore_view(ident):
+def webcapture_underscore_view(ident: str) -> AnyResponse:
return redirect("/webcapture/{}".format(ident))
@app.route("/webcapture/<string(length=26):ident>/metadata", methods=["GET"])
-def webcapture_view_metadata(ident):
+def webcapture_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("webcapture", ident, "entity_view_metadata.html")
@app.route("/webcapture/rev/<uuid:revision_id>", methods=["GET"])
-def webcapture_revision_view(revision_id):
+def webcapture_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("webcapture", str(revision_id), "webcapture_view.html")
@app.route("/webcapture/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def webcapture_revision_view_metadata(revision_id):
+def webcapture_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"webcapture", str(revision_id), "entity_view_metadata.html"
)
@app.route("/editgroup/<editgroup_id>/webcapture/<string(length=26):ident>", methods=["GET"])
-def webcapture_editgroup_view(editgroup_id, ident):
+def webcapture_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "webcapture", ident, "webcapture_view.html"
)
@@ -527,72 +534,72 @@ def webcapture_editgroup_view(editgroup_id, ident):
@app.route(
"/editgroup/<editgroup_id>/webcapture/<string(length=26):ident>/metadata", methods=["GET"]
)
-def webcapture_editgroup_view_metadata(editgroup_id, ident):
+def webcapture_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "webcapture", ident, "entity_view_metadata.html"
)
@app.route("/release/<string(length=26):ident>", methods=["GET"])
-def release_view(ident):
+def release_view(ident: str) -> AnyResponse:
return generic_entity_view("release", ident, "release_view.html")
@app.route("/release_<string(length=26):ident>", methods=["GET"])
-def release_underscore_view(ident):
+def release_underscore_view(ident: str) -> AnyResponse:
return redirect("/release/{}".format(ident))
@app.route("/release/<string(length=26):ident>/contribs", methods=["GET"])
-def release_view_contribs(ident):
+def release_view_contribs(ident: str) -> AnyResponse:
return generic_entity_view("release", ident, "release_view_contribs.html")
@app.route("/release/<string(length=26):ident>/references", methods=["GET"])
-def release_view_references(ident):
+def release_view_references(ident: str) -> AnyResponse:
return generic_entity_view("release", ident, "release_view_references.html")
@app.route("/release/<string(length=26):ident>/metadata", methods=["GET"])
-def release_view_metadata(ident):
+def release_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("release", ident, "entity_view_metadata.html")
@app.route("/release/rev/<uuid:revision_id>", methods=["GET"])
-def release_revision_view(revision_id):
+def release_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("release", str(revision_id), "release_view.html")
@app.route("/release/rev/<uuid:revision_id>/contribs", methods=["GET"])
-def release_revision_view_contribs(revision_id):
+def release_revision_view_contribs(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"release", str(revision_id), "release_view_contribs.html"
)
@app.route("/release/rev/<uuid:revision_id>/references", methods=["GET"])
-def release_revision_view_references(revision_id):
+def release_revision_view_references(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"release", str(revision_id), "release_view_references.html"
)
@app.route("/release/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def release_revision_view_metadata(revision_id):
+def release_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view(
"release", str(revision_id), "entity_view_metadata.html"
)
@app.route("/editgroup/<editgroup_id>/release/<string(length=26):ident>", methods=["GET"])
-def release_editgroup_view(editgroup_id, ident):
+def release_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(editgroup_id, "release", ident, "release_view.html")
@app.route(
"/editgroup/<editgroup_id>/release/<string(length=26):ident>/contribs", methods=["GET"]
)
-def release_editgroup_view_contribs(editgroup_id, ident):
+def release_editgroup_view_contribs(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "release", ident, "release_view_contribs.html"
)
@@ -601,7 +608,7 @@ def release_editgroup_view_contribs(editgroup_id, ident):
@app.route(
"/editgroup/<editgroup_id>/release/<string(length=26):ident>/references", methods=["GET"]
)
-def release_editgroup_view_references(editgroup_id, ident):
+def release_editgroup_view_references(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "release", ident, "release_view_references.html"
)
@@ -610,44 +617,44 @@ def release_editgroup_view_references(editgroup_id, ident):
@app.route(
"/editgroup/<editgroup_id>/release/<string(length=26):ident>/metadata", methods=["GET"]
)
-def release_editgroup_view_metadata(editgroup_id, ident):
+def release_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "release", ident, "entity_view_metadata.html"
)
@app.route("/work/<string(length=26):ident>", methods=["GET"])
-def work_view(ident):
+def work_view(ident: str) -> AnyResponse:
return generic_entity_view("work", ident, "work_view.html")
@app.route("/work_<string(length=26):ident>", methods=["GET"])
-def work_underscore_view(ident):
+def work_underscore_view(ident: str) -> AnyResponse:
return redirect("/work/{}".format(ident))
@app.route("/work/<string(length=26):ident>/metadata", methods=["GET"])
-def work_view_metadata(ident):
+def work_view_metadata(ident: str) -> AnyResponse:
return generic_entity_view("work", ident, "entity_view_metadata.html")
@app.route("/work/rev/<uuid:revision_id>", methods=["GET"])
-def work_revision_view(revision_id):
+def work_revision_view(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("work", str(revision_id), "work_view.html")
@app.route("/work/rev/<uuid:revision_id>/metadata", methods=["GET"])
-def work_revision_view_metadata(revision_id):
+def work_revision_view_metadata(revision_id: str) -> AnyResponse:
return generic_entity_revision_view("work", str(revision_id), "entity_view_metadata.html")
@app.route("/editgroup/<editgroup_id>/work/<string(length=26):ident>", methods=["GET"])
-def work_editgroup_view(editgroup_id, ident):
+def work_editgroup_view(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(editgroup_id, "work", ident, "work_view.html")
@app.route("/editgroup/<editgroup_id>/work/<string(length=26):ident>/metadata", methods=["GET"])
-def work_editgroup_view_metadata(editgroup_id, ident):
+def work_editgroup_view_metadata(editgroup_id: str, ident: str) -> AnyResponse:
return generic_editgroup_entity_view(
editgroup_id, "work", ident, "entity_view_metadata.html"
)
@@ -657,7 +664,7 @@ def work_editgroup_view_metadata(editgroup_id, ident):
@app.route("/editgroup/<string(length=26):ident>", methods=["GET"])
-def editgroup_view(ident):
+def editgroup_view(ident: str) -> AnyResponse:
try:
eg = api.get_editgroup(str(ident))
eg.editor = api.get_editor(eg.editor_id)
@@ -684,7 +691,7 @@ def editgroup_view(ident):
@app.route("/editgroup/<string(length=26):ident>/annotation", methods=["POST"])
@login_required
-def editgroup_create_annotation(ident):
+def editgroup_create_annotation(ident: str) -> AnyResponse:
if not app.testing:
app.csrf.protect()
comment_markdown = request.form.get("comment_markdown")
@@ -710,7 +717,7 @@ def editgroup_create_annotation(ident):
@app.route("/editgroup/<string(length=26):ident>/accept", methods=["POST"])
@login_required
-def editgroup_accept(ident):
+def editgroup_accept(ident: str) -> AnyResponse:
if not app.testing:
app.csrf.protect()
# on behalf of user...
@@ -728,7 +735,7 @@ def editgroup_accept(ident):
@app.route("/editgroup/<string(length=26):ident>/unsubmit", methods=["POST"])
@login_required
-def editgroup_unsubmit(ident):
+def editgroup_unsubmit(ident: str) -> AnyResponse:
if not app.testing:
app.csrf.protect()
# on behalf of user...
@@ -746,7 +753,7 @@ def editgroup_unsubmit(ident):
@app.route("/editgroup/<string(length=26):ident>/submit", methods=["POST"])
@login_required
-def editgroup_submit(ident):
+def editgroup_submit(ident: str) -> AnyResponse:
if not app.testing:
app.csrf.protect()
# on behalf of user...
@@ -763,7 +770,7 @@ def editgroup_submit(ident):
@app.route("/editor/<string(length=26):ident>", methods=["GET"])
-def editor_view(ident):
+def editor_view(ident: str) -> AnyResponse:
try:
entity = api.get_editor(ident)
except ApiException as ae:
@@ -772,7 +779,7 @@ def editor_view(ident):
@app.route("/editor/<string(length=26):ident>/editgroups", methods=["GET"])
-def editor_editgroups(ident):
+def editor_editgroups(ident: str) -> AnyResponse:
try:
editor = api.get_editor(ident)
editgroups = api.get_editor_editgroups(ident, limit=50)
@@ -785,7 +792,7 @@ def editor_editgroups(ident):
@app.route("/editor/<string(length=26):ident>/annotations", methods=["GET"])
-def editor_annotations(ident):
+def editor_annotations(ident: str) -> AnyResponse:
try:
editor = api.get_editor(ident)
annotations = api.get_editor_annotations(ident, limit=50)
@@ -795,7 +802,7 @@ def editor_annotations(ident):
@app.route("/u/<string:username>", methods=["GET", "HEAD"])
-def editor_username_redirect(username):
+def editor_username_redirect(username: str) -> AnyResponse:
try:
editor = api.lookup_editor(username=username)
except ApiException as ae:
@@ -804,7 +811,7 @@ def editor_username_redirect(username):
@app.route("/changelog", methods=["GET"])
-def changelog_view():
+def changelog_view() -> AnyResponse:
try:
# limit = int(request.args.get('limit', 10))
entries = api.get_changelog() # TODO: expand="editors"
@@ -814,7 +821,7 @@ def changelog_view():
@app.route("/changelog/<int:index>", methods=["GET"])
-def changelog_entry_view(index):
+def changelog_entry_view(index: int) -> AnyResponse:
try:
entry = api.get_changelog_entry(int(index))
entry.editgroup.editor = api.get_editor(entry.editgroup.editor_id)
@@ -827,7 +834,7 @@ def changelog_entry_view(index):
@app.route("/reviewable", methods=["GET"])
-def reviewable_view():
+def reviewable_view() -> AnyResponse:
try:
# limit = int(request.args.get('limit', 10))
entries = api.get_editgroups_reviewable(expand="editors")
@@ -837,7 +844,7 @@ def reviewable_view():
@app.route("/release/<string(length=26):ident>/save", methods=["GET", "POST"])
-def release_save(ident):
+def release_save(ident: str) -> AnyResponse:
form = SavePaperNowForm()
@@ -902,7 +909,7 @@ def release_save(ident):
@app.route("/search", methods=["GET", "POST"])
-def generic_search():
+def generic_search() -> AnyResponse:
if "q" not in request.args.keys():
return redirect("/release/search")
query = request.args.get("q").strip()
@@ -932,7 +939,7 @@ def generic_search():
@app.route("/release/search", methods=["GET", "POST"])
-def release_search():
+def release_search() -> AnyResponse:
if "q" not in request.args.keys():
return render_template("release_search.html", query=ReleaseQuery(), found=None)
@@ -960,7 +967,7 @@ def release_search():
@app.route("/container/search", methods=["GET", "POST"])
-def container_search():
+def container_search() -> AnyResponse:
if "q" not in request.args.keys():
return render_template("container_search.html", query=GenericQuery(), found=None)
@@ -977,7 +984,7 @@ def container_search():
@app.route("/coverage/search", methods=["GET", "POST"])
-def coverage_search():
+def coverage_search() -> AnyResponse:
if "q" not in request.args.keys():
return render_template(
@@ -1032,7 +1039,7 @@ def coverage_search():
)
-def get_changelog_stats():
+def get_changelog_stats() -> Dict[str, Any]:
stats = {}
latest_changelog = api.get_changelog(limit=1)[0]
stats["changelog"] = {
@@ -1045,7 +1052,7 @@ def get_changelog_stats():
@app.route("/stats", methods=["GET"])
-def stats_page():
+def stats_page() -> AnyResponse:
try:
stats = get_elastic_entity_stats()
stats.update(get_changelog_stats())
@@ -1060,7 +1067,7 @@ def stats_page():
@app.route("/stats.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def stats_json():
+def stats_json() -> AnyResponse:
try:
stats = get_elastic_entity_stats()
stats.update(get_changelog_stats())
@@ -1072,7 +1079,7 @@ def stats_json():
@app.route("/container/issnl/<issnl>/stats.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_issnl_stats(issnl):
+def container_issnl_stats(issnl: str) -> AnyResponse:
if not (len(issnl) == 9 and issnl[4] == "-"):
abort(400, "Not a valid ISSN-L: {}".format(issnl))
try:
@@ -1089,7 +1096,7 @@ def container_issnl_stats(issnl):
@app.route("/container/<string(length=26):ident>/stats.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_stats(ident):
+def container_ident_stats(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1106,7 +1113,7 @@ def container_ident_stats(ident):
"/container/<string(length=26):ident>/ia_coverage_years.json", methods=["GET", "OPTIONS"]
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_ia_coverage_years_json(ident):
+def container_ident_ia_coverage_years_json(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1124,7 +1131,7 @@ def container_ident_ia_coverage_years_json(ident):
"/container/<string(length=26):ident>/ia_coverage_years.svg", methods=["GET", "OPTIONS"]
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_ia_coverage_years_svg(ident):
+def container_ident_ia_coverage_years_svg(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1141,7 +1148,7 @@ def container_ident_ia_coverage_years_svg(ident):
"/container/<string(length=26):ident>/preservation_by_year.json", methods=["GET", "OPTIONS"]
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_preservation_by_year_json(ident):
+def container_ident_preservation_by_year_json(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1159,7 +1166,7 @@ def container_ident_preservation_by_year_json(ident):
"/container/<string(length=26):ident>/preservation_by_year.svg", methods=["GET", "OPTIONS"]
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_preservation_by_year_svg(ident):
+def container_ident_preservation_by_year_svg(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1181,7 +1188,7 @@ def container_ident_preservation_by_year_svg(ident):
methods=["GET", "OPTIONS"],
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_preservation_by_volume_json(ident):
+def container_ident_preservation_by_volume_json(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1199,7 +1206,7 @@ def container_ident_preservation_by_volume_json(ident):
methods=["GET", "OPTIONS"],
)
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def container_ident_preservation_by_volume_svg(ident):
+def container_ident_preservation_by_volume_svg(ident: str) -> AnyResponse:
try:
container = api.get_container(ident)
except ApiException as ae:
@@ -1216,7 +1223,7 @@ def container_ident_preservation_by_volume_svg(ident):
@app.route("/release/<string(length=26):ident>.bib", methods=["GET"])
-def release_bibtex(ident):
+def release_bibtex(ident: str) -> AnyResponse:
try:
entity = api.get_release(ident)
except ApiException as ae:
@@ -1227,7 +1234,7 @@ def release_bibtex(ident):
@app.route("/release/<string(length=26):ident>/citeproc", methods=["GET"])
-def release_citeproc(ident):
+def release_citeproc(ident: str) -> AnyResponse:
style = request.args.get("style", "harvard1")
is_html = request.args.get("html", False)
if is_html and is_html.lower() in ("yes", "1", "true", "y", "t"):
@@ -1254,7 +1261,7 @@ def release_citeproc(ident):
@app.route("/health.json", methods=["GET", "OPTIONS"])
@crossdomain(origin="*", headers=["access-control-allow-origin", "Content-Type"])
-def health_json():
+def health_json() -> AnyResponse:
return jsonify({"ok": True})
@@ -1262,7 +1269,7 @@ def health_json():
@app.route("/auth/login")
-def login():
+def login() -> AnyResponse:
# show the user a list of login options
if not priv_api:
app.log.warn(
@@ -1272,7 +1279,7 @@ def login():
@app.route("/auth/ia/login", methods=["GET", "POST"])
-def ia_xauth_login():
+def ia_xauth_login() -> AnyResponse:
if "email" in request.form:
# if a login attempt...
return handle_ia_xauth(request.form.get("email"), request.form.get("password"))
@@ -1281,7 +1288,7 @@ def ia_xauth_login():
@app.route("/auth/token_login", methods=["GET", "POST"])
-def token_login():
+def token_login() -> AnyResponse:
# show the user a list of login options
if "token" in request.args:
return handle_token_login(request.args.get("token"))
@@ -1292,7 +1299,7 @@ def token_login():
@app.route("/auth/change_username", methods=["POST"])
@login_required
-def change_username():
+def change_username() -> AnyResponse:
if not app.testing:
app.csrf.protect()
# show the user a list of login options
@@ -1315,7 +1322,7 @@ def change_username():
@app.route("/auth/create_token", methods=["POST"])
@login_required
-def create_auth_token():
+def create_auth_token() -> AnyResponse:
if not app.testing:
app.csrf.protect()
@@ -1346,14 +1353,14 @@ def create_auth_token():
@app.route("/auth/logout")
-def logout():
+def logout() -> AnyResponse:
handle_logout()
return render_template("auth_logout.html")
@app.route("/auth/account")
@login_required
-def auth_account():
+def auth_account() -> AnyResponse:
# auth check on account page
user_api = auth_api(session["api_token"])
resp = user_api.auth_check()
@@ -1365,7 +1372,7 @@ def auth_account():
@app.route("/auth/wikipedia/auth")
-def wp_oauth_rewrite():
+def wp_oauth_rewrite() -> AnyResponse:
"""
This is a dirty hack to rewrite '/auth/wikipedia/auth' to '/auth/wikipedia/oauth-callback'
"""
@@ -1376,7 +1383,7 @@ def wp_oauth_rewrite():
@app.route("/auth/wikipedia/finish-login")
-def wp_oauth_finish_login():
+def wp_oauth_finish_login() -> AnyResponse:
wp_username = mwoauth.get_current_user(cached=True)
assert wp_username
return handle_wmoauth(wp_username)
@@ -1386,33 +1393,33 @@ def wp_oauth_finish_login():
@app.errorhandler(404)
-def page_not_found(e):
+def page_not_found(e: Exception) -> AnyResponse:
return render_template("404.html"), 404
@app.errorhandler(401)
@app.errorhandler(403)
-def page_not_authorized(e):
+def page_not_authorized(e: Exception) -> AnyResponse:
return render_template("403.html"), 403
@app.errorhandler(405)
-def page_method_not_allowed(e):
+def page_method_not_allowed(e: Exception) -> AnyResponse:
return render_template("405.html"), 405
@app.errorhandler(400)
-def page_bad_request(e):
+def page_bad_request(e: Exception) -> AnyResponse:
return render_template("400.html", err=e), 400
@app.errorhandler(409)
-def page_edit_conflict(e):
+def page_edit_conflict(e: Exception) -> AnyResponse:
return render_template("409.html"), 409
@app.errorhandler(500)
-def page_server_error(e):
+def page_server_error(e: Exception) -> AnyResponse:
app.log.error(e)
return render_template("500.html"), 500
@@ -1420,13 +1427,13 @@ def page_server_error(e):
@app.errorhandler(502)
@app.errorhandler(503)
@app.errorhandler(504)
-def page_server_down(e):
+def page_server_down(e: Exception) -> AnyResponse:
app.log.error(e)
return render_template("503.html"), 503
@app.errorhandler(ApiException)
-def page_fatcat_api_error(ae):
+def page_fatcat_api_error(ae: ApiException) -> AnyResponse:
"""
Generic error handler for fatcat API problems. With this error handler,
don't need to explicitly catch API exceptions: they should get caught and
@@ -1450,7 +1457,7 @@ def page_fatcat_api_error(ae):
@app.errorhandler(ApiValueError)
-def page_fatcat_api_value_error(ae):
+def page_fatcat_api_value_error(ae: ApiValueError) -> AnyResponse:
ae.status = 400
ae.error_name = "ValueError"
ae.message = str(ae)
@@ -1458,27 +1465,27 @@ def page_fatcat_api_value_error(ae):
@app.errorhandler(CSRFError)
-def page_csrf_error(e):
+def page_csrf_error(e: CSRFError) -> AnyResponse:
return render_template("csrf_error.html", reason=e.description), 400
@app.route("/", methods=["GET"])
-def page_home():
+def page_home() -> AnyResponse:
return render_template("home.html")
@app.route("/about", methods=["GET"])
-def page_about():
+def page_about() -> AnyResponse:
return render_template("about.html")
@app.route("/rfc", methods=["GET"])
-def page_rfc():
+def page_rfc() -> AnyResponse:
return render_template("rfc.html")
@app.route("/robots.txt", methods=["GET"])
-def page_robots_txt():
+def page_robots_txt() -> AnyResponse:
if app.config["FATCAT_DOMAIN"] == "fatcat.wiki":
robots_path = "robots.txt"
else:
@@ -1489,7 +1496,7 @@ def page_robots_txt():
@app.route("/sitemap.xml", methods=["GET"])
-def page_sitemap_xml():
+def page_sitemap_xml() -> AnyResponse:
return send_from_directory(
os.path.join(app.root_path, "static"), "sitemap.xml", mimetype="text/xml"
)
diff --git a/python/fatcat_web/search.py b/python/fatcat_web/search.py
index 5e758fd0..06660cd6 100644
--- a/python/fatcat_web/search.py
+++ b/python/fatcat_web/search.py
@@ -6,7 +6,7 @@ the formal API)
import datetime
import sys
from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
import elasticsearch
import elasticsearch_dsl.response
@@ -33,8 +33,8 @@ class ReleaseQuery:
container_id: Optional[str] = None
recent: bool = False
- @classmethod
- def from_args(cls, args) -> "ReleaseQuery":
+ @staticmethod
+ def from_args(args: Dict[str, Any]) -> "ReleaseQuery":
query_str = args.get("q") or "*"
@@ -66,8 +66,8 @@ class GenericQuery:
limit: Optional[int] = None
offset: Optional[int] = None
- @classmethod
- def from_args(cls, args) -> "GenericQuery":
+ @staticmethod
+ def from_args(args: Dict[str, Any]) -> "GenericQuery":
query_str = args.get("q")
if not query_str:
query_str = "*"
@@ -154,7 +154,7 @@ def wrap_es_execution(search: Search) -> Any:
return resp
-def agg_to_dict(agg) -> dict:
+def agg_to_dict(agg: Any) -> Dict[str, Any]:
"""
Takes a simple term aggregation result (with buckets) and returns a simple
dict with keys as terms and counts as values. Includes an extra value
@@ -289,7 +289,7 @@ def do_release_search(query: ReleaseQuery, deep_page_limit: int = 2000) -> Searc
)
-def get_elastic_container_random_releases(ident: str, limit=5) -> List[Dict[str, Any]]:
+def get_elastic_container_random_releases(ident: str, limit: int = 5) -> List[Dict[str, Any]]:
"""
Returns a list of releases from the container.
"""
@@ -453,8 +453,12 @@ def get_elastic_search_coverage(query: ReleaseQuery) -> dict:
def get_elastic_container_stats(
- ident, issnl=None, es_client=None, es_index=None, merge_shadows=None
-):
+ ident: str,
+ issnl: Optional[str] = None,
+ es_client: Optional[elasticsearch.Elasticsearch] = None,
+ es_index: Optional[str] = None,
+ merge_shadows: Optional[bool] = None,
+) -> Dict[str, Any]:
"""
Returns dict:
ident
@@ -535,7 +539,7 @@ def get_elastic_container_stats(
return stats
-def get_elastic_container_histogram_legacy(ident) -> List:
+def get_elastic_container_histogram_legacy(ident: str) -> List[Tuple[int, bool, int]]:
"""
Fetches a stacked histogram of {year, in_ia}. This is for the older style
of coverage graph (SVG or JSON export). This function should be DEPRECATED
@@ -603,7 +607,7 @@ def get_elastic_container_histogram_legacy(ident) -> List:
return vals
-def get_elastic_preservation_by_year(query) -> List[dict]:
+def get_elastic_preservation_by_year(query: ReleaseQuery) -> List[Dict[str, Any]]:
"""
Fetches a stacked histogram of {year, preservation}.
@@ -685,7 +689,7 @@ def get_elastic_preservation_by_year(query) -> List[dict]:
return sorted(year_dicts.values(), key=lambda x: x["year"])
-def get_elastic_preservation_by_date(query) -> List[dict]:
+def get_elastic_preservation_by_date(query: ReleaseQuery) -> List[dict]:
"""
Fetches a stacked histogram of {date, preservation}.