aboutsummaryrefslogtreecommitdiff
path: root/gn2/wqflask/oauth2/request_utils.py
blob: 456aba2be25e7936d29fb04421c1d04b03127d6d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
"""General request utilities"""
from typing import Optional, Callable
from urllib.parse import urljoin, urlparse
from requests.exceptions import JSONDecodeError

import simplejson
from flask import (
    flash, request, url_for, redirect, Response, render_template,
    current_app as app)

from gn2.wqflask.external_errors import ExternalRequestError

from .client import (
    SCOPE, oauth2_get, authserver_uri, oauth2_clientid, oauth2_clientsecret)

def authserver_authorise_uri():
    req_baseurl = urlparse(request.base_url, scheme=request.scheme)
    host_uri = f"{req_baseurl.scheme}://{req_baseurl.netloc}/"
    return urljoin(
        authserver_uri(),
        "auth/authorise?response_type=code"
        f"&client_id={oauth2_clientid()}"
        f"&redirect_uri={urljoin(host_uri, 'oauth2/code')}")


def user_details():
    return oauth2_get("auth/user/").either(
        lambda err: {},
        lambda usr_dets: usr_dets)

def process_error(error: Response,
                  message: str=("Requested endpoint was not found on the API "
                                "server.")
                  ) -> dict:
    if error.status_code in range(400, 500):
        try:
            err = error.json()
            msg = err.get(
                "error_message", err.get("error_description", f"{error.reason}"))
        except simplejson.errors.JSONDecodeError as _jde:
            msg = message
        return {
            "error": error.reason,
            "error_message": msg,
            "error_description": msg,
            "status_code": error.status_code
        }
    try:
        return {**error.json(), "status_code": error.status_code}
    except JSONDecodeError as exc:
        raise ExternalRequestError(
            error.url,
            exc,
            f"Could not parse error record into JSON:\n\n{error.content}")

def request_error(response):
    app.logger.error(f"{response}: {response.url} [{response.status_code}]")
    return render_template("oauth2/request_error.html", response=response)

def handle_error(redirect_uri: Optional[str] = None, **kwargs):
    def __handler__(error):
        error_json = process_error(error)# error.json()
        msg = error_json.get(
            "error_message", error_json.get(
                "error_description", "undefined error"))
        flash(f"{error_json['error']}: {msg}.",
              "alert-danger")
        if "response_handlers" in kwargs:
            for handler in kwargs["response_handlers"]:
                handler(response)
        if redirect:
            return redirect(url_for(redirect_uri, **kwargs))

    return __handler__

def handle_success(
        success_msg: str, redirect_uri: Optional[str] = None, **kwargs):
    def __handler__(response):
        flash(f"Success: {success_msg}.", "alert-success")
        if "response_handlers" in kwargs:
            for handler in kwargs["response_handlers"]:
                handler(response)
        if redirect:
            return redirect(url_for(redirect_uri, **kwargs))

    return __handler__

def flash_error(error):
    flash(f"{error['error']}: {error['error_description']}", "alert-danger")

def flash_success(success):
    flash(f"{success['description']}", "alert-success")

def with_flash_error(response) -> Callable:
    def __err__(err) -> Response:
        error = process_error(err)
        flash(f"{error['status_code']} {error['error']}: "
              f"{error['error_description']}",
              "alert-danger")
        return response
    return __err__

def with_flash_success(response) -> Callable:
    def __succ__(msg) -> Response:
        flash(f"Success: {msg['message']}", "alert-success")
        return response
    return __succ__