aboutsummaryrefslogtreecommitdiff
path: root/uploader/monadic_requests.py
blob: aa34951a9140b05c807441198884e56d45d11430 (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
"""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)