aboutsummaryrefslogtreecommitdiff
path: root/uploader/oauth2
diff options
context:
space:
mode:
Diffstat (limited to 'uploader/oauth2')
-rw-r--r--uploader/oauth2/client.py16
1 files changed, 9 insertions, 7 deletions
diff --git a/uploader/oauth2/client.py b/uploader/oauth2/client.py
index 28464a3..09e165a 100644
--- a/uploader/oauth2/client.py
+++ b/uploader/oauth2/client.py
@@ -1,4 +1,5 @@
"""OAuth2 client utilities."""
+import json
from datetime import datetime, timedelta
from urllib.parse import urljoin, urlparse
@@ -36,7 +37,7 @@ def oauth2_clientsecret():
def __make_token_validator__(keys: KeySet):
"""Make a token validator function."""
- def __validator__(token: str):
+ def __validator__(token: dict):
for key in keys.keys:
try:
jwt.decode(token["access_token"], key)
@@ -110,6 +111,7 @@ def user_logged_in():
def authserver_authorise_uri():
+ """Build up the authorisation URI."""
req_baseurl = urlparse(request.base_url, scheme=request.scheme)
host_uri = f"{req_baseurl.scheme}://{req_baseurl.netloc}/"
return urljoin(
@@ -122,7 +124,7 @@ def authserver_authorise_uri():
def __no_token__(_err) -> Left:
"""Handle situation where request is attempted with no token."""
resp = requests.models.Response()
- resp._content = json.dumps({
+ resp._content = json.dumps({#pylint: disable=[protected-access]
"error": "AuthenticationError",
"error-description": ("You need to authenticate to access requested "
"information.")}).encode("utf-8")
@@ -132,7 +134,7 @@ def __no_token__(_err) -> Left:
def oauth2_get(url, **kwargs) -> Either:
"""Do a get request to the authentication/authorisation server."""
- def __get__(token) -> Either:
+ def __get__(_token) -> Either:
_uri = urljoin(authserver_uri(), url)
try:
resp = oauth2_client().get(
@@ -147,7 +149,7 @@ def oauth2_get(url, **kwargs) -> Either:
if resp.status_code in mrequests.SUCCESS_CODES:
return Right(resp.json())
return Left(resp)
- except Exception as exc:
+ except Exception as exc:#pylint: disable=[broad-except]
app.logger.error("Error retriving data from auth server: (GET %s)",
_uri,
exc_info=True)
@@ -155,9 +157,9 @@ def oauth2_get(url, **kwargs) -> Either:
return session.user_token().either(__no_token__, __get__)
-def oauth2_post(url, data=None, json=None, **kwargs):
+def oauth2_post(url, data=None, json=None, **kwargs):#pylint: disable=[redefined-outer-name]
"""Do a POST request to the authentication/authorisation server."""
- def __post__(token) -> Either:
+ def __post__(_token) -> Either:
_uri = urljoin(authserver_uri(), url)
_headers = ({
**kwargs.get("headers", {}),
@@ -179,7 +181,7 @@ def oauth2_post(url, data=None, json=None, **kwargs):
if resp.status_code in mrequests.SUCCESS_CODES:
return Right(resp.json())
return Left(resp)
- except Exception as exc:
+ except Exception as exc:#pylint: disable=[broad-except]
app.logger.error("Error retriving data from auth server: (POST %s)",
_uri,
exc_info=True)