aboutsummaryrefslogtreecommitdiff
path: root/uploader/monadic_requests.py
blob: c492df588f8d66ba84c3b5a7a53b5723a7ce9cb4 (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
"""Wrap requests functions with monads."""
import traceback
from typing import Union, Optional, Callable

import requests
from requests.models import Response
from pymonad.either import Left, Right, Either
from flask import (flash,
                   request,
                   redirect,
                   render_template,
                   current_app as app,
                   escape as flask_escape)

# HTML Status codes indicating a successful request.
SUCCESS_CODES = (200, 201, 202, 203, 204, 205, 206, 207, 208, 226)

# Possible error(s) that can be encontered while attempting to do a request.
PossibleError = Union[Response, Exception]


def make_error_handler(
        redirect_to: Optional[Response] = None,
        cleanup_thunk: Callable = lambda *args: None
) -> Callable[[PossibleError], Response]:
    """
    Build a function to gracefully handle errors encountered while doing
    requests.

    :rtype: Callable
    """
    redirect_to = redirect_to or redirect(request.url)
    def __handler__(resp_or_exc: PossibleError) -> Response:
        cleanup_thunk()
        if issubclass(type(resp_or_exc), Exception):
            # Is an exception!
            return render_template(
                "unhandled_exception.html",
                trace=traceback.format_exception(resp_or_exc))
        if isinstance(resp_or_exc, Response):
            flash("The authorisation server responded with "
                  f"({flask_escape(resp_or_exc.status_code)}, "
                  f"{flask_escape(resp_or_exc.reason)}) for the request to "
                  f"'{flask_escape(resp_or_exc.request.url)}'",
                  "alert-danger")
            return redirect_to

        flash("Unspecified error!", "alert-danger")
        app.logger.debug("Error (%s): %s", type(resp_or_exc), resp_or_exc)
        return redirect_to
    return __handler__


def get(url, params=None, **kwargs) -> Either:
    """
    A wrapper around `requests.get` function.

    Takes the same arguments as `requests.get`.

    :rtype: pymonad.either.Either
    """
    try:
        resp = requests.get(url, params=params, **kwargs)
        if resp.status_code in SUCCESS_CODES:
            return Right(resp.json())
        return Left(resp)
    except requests.exceptions.RequestException as exc:
        return Left(exc)


def post(url, data=None, json=None, **kwargs) -> Either:
    """
    A wrapper around `requests.post` function.

    Takes the same arguments as `requests.post`.

    :rtype: pymonad.either.Either
    """
    try:
        resp = requests.post(url, data=data, json=json, **kwargs)
        if resp.status_code in SUCCESS_CODES:
            return Right(resp.json())
        return Left(resp)
    except requests.exceptions.RequestException as exc:
        return Left(exc)


def make_either_error_handler(msg):
    """Make generic error handler for pymonads Either objects."""
    def __fail__(error):
        if issubclass(type(error), Exception):
            app.logger.debug("\n\n%s (Exception)\n\n", msg, exc_info=True)
            raise error
        if issubclass(type(error), Response):
            try:
                _data = error.json()
            except Exception as _exc:
                raise Exception(error.content) from _exc
            raise Exception(_data)

        app.logger.debug("\n\n%s\n\n", msg)
        raise Exception(error)

    return __fail__