StarfleetCommander/restRequests.py at master · jonathancutting/StarfleetCommander · GitHub

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
import requests     # built-in REST request handling
import logging      # built-in Python logging

logger = logging.getLogger(__name__)    # set module-level logger object

def sendGetRequest(url:str, params:dict, s:requests.Session) -> requests.Response:
    '''
    Sends a GET request to the specified URL and returns response object.
    If URL is invalid, or request is unsuccessful, an an error is raised.

    Args:
        url (str): location to send the GET request
        params (dict, optional): headers to send with the request
        s (requests.Session, optional): Session object to manage headers

    Returns:
        requests.response: HTML response object
    '''
    if not url: raise ValueError("URL cannot be empty.")
    try:
        if not s:
            r = requests.get(url, params=params)
        else:
            r = s.get(url, params=params)
        r.raise_for_status() # Raise exception for 4xx and 5xx status codes.
        return r
    except:
        raise

def sendPostRequest(url:str, body:dict, hdrs:dict, s:requests.Session) -> requests.Response:
    '''
    Sends a POST request to the specified URL and returns the response object.
    If URL is invalid, or request is unsuccessful, then an error is raised.

    Args:
        url (str): location to send the POST request
        body (dict): information to send to the remote server
        hdrs (dict): header information to include in the request
        s (requests.Session, optional): Session object to manage headers

    Returns:
        requests.response: HTML response object
    '''
    if not url: raise ValueError("URL cannot by empty.")
    try:
        if not s:
            r = requests.post(url, data=body, headers=hdrs)
        else:
            r = s.post(url, data=body, headers=hdrs)
        r.raise_for_status() # Raise exception for 4xx and 5xx status codes.
        return r
    except:
        raise

def sendRequest(req:dict) -> requests.Response:
    '''
    Takes a generic request object and interprets it to determine which REST
    function should be called. Checks first that request dictionary is properly
    formed. Any errors raised by REST functions are passed to caller.

    Args:
        req (dict): dictionary containing the following keys:
            url (str): the URL of the request target
            sess (requests.Session): the current requests session object
            hdr (dict): dictionary containing the request headers. If missing,
                        an empty dictionary will be created for it.
            body (dict, opt): dictionary containing the request body, optional
                              if making a GET request

    Returns:
        requests.Response: HTML response object
    '''

    # Raise error, if necessary members are missing.
    if not "url" in req:
        raise ValueError("URL is missing from request!")
    if not "sess" in req:
        raise ValueError("Session object is missing from request!")

    # Prepare empty header dictionary, if missing.
    if not "hdr" in req:
        req["hdr"] = {}

    # Detect appropriate request function and make call. Propogate any errors to caller.
    try:
        if not "body" in req:
            r = sendGetRequest(req["url"], req["hdr"], req["sess"])
        else:
            r = sendPostRequest(req["url"], req["body"], req["hdr"], req["sess"])
    except:
        raise

    return r