lms.backend.canvas.common

  1import datetime
  2import http
  3import re
  4import typing
  5
  6import edq.net.request
  7import edq.util.json
  8import edq.util.time
  9import requests
 10
 11DEFAULT_PAGE_SIZE: int = 95
 12HEADER_LINK: str = 'Link'
 13
 14def fetch_next_canvas_link(response: requests.Response) -> typing.Union[str, None]:
 15    """
 16    Fetch the Canvas-style next link within the headers.
 17    If there is no next link, return None.
 18    """
 19
 20    headers = response.headers
 21
 22    if (HEADER_LINK not in headers):
 23        return None
 24
 25    links = headers[HEADER_LINK].split(',')
 26    for link in links:
 27        parts = link.split(';')
 28        if (len(parts) != 2):
 29            continue
 30
 31        if (parts[1].strip() != 'rel="next"'):
 32            continue
 33
 34        return str(parts[0].strip().strip('<>'))
 35
 36    return None
 37
 38def make_request(
 39        method: str,
 40        url: str,
 41        raise_on_404: bool = False,
 42        json: bool = True,
 43        **kwargs: typing.Any) -> typing.Union[typing.Any, None]:
 44    """ Make a single Canvas request and return the decoded JSON body. """
 45
 46    try:
 47        _, body_text = edq.net.request.make_request(method, url, **kwargs)
 48    except requests.HTTPError as ex:
 49        if (raise_on_404 or (ex.response is None) or (ex.response.status_code != http.HTTPStatus.NOT_FOUND)):
 50            raise ex
 51
 52        return None
 53
 54    if (not json):
 55        return body_text
 56
 57    return edq.util.json.loads(body_text, strict = True)
 58
 59def make_get_request(url: str, **kwargs: typing.Any) -> typing.Union[typing.Any, None]:
 60    """ Make a single Canvas GET request. """
 61
 62    return make_request('GET', url, **kwargs)
 63
 64def make_post_request(url: str, **kwargs: typing.Any) -> typing.Union[typing.Any, None]:
 65    """ Make a single Canvas POST request. """
 66
 67    return make_request('POST', url, **kwargs)
 68
 69def make_delete_request(url: str, **kwargs: typing.Any) -> typing.Union[typing.Any, None]:
 70    """ Make a single Canvas DELETE request. """
 71
 72    return make_request('DELETE', url, **kwargs)
 73
 74def make_get_request_list(
 75        url: str,
 76        headers: typing.Dict[str, typing.Any],
 77        data: typing.Union[typing.Dict[str, typing.Any], None] = None,
 78        raise_on_404: bool = False,
 79        ) -> typing.Union[typing.List[typing.Dict[str, typing.Any]], None]:
 80    """ Repeatedly call make_get_request() (using a JSON body and next link) until there are no more results. """
 81
 82    output: typing.List[typing.Dict[str, typing.Any]] = []
 83
 84    next_url: typing.Union[str, None] = url
 85
 86    while (next_url is not None):
 87        try:
 88            response, body_text = edq.net.request.make_get(next_url, headers = headers, data = data)
 89        except requests.HTTPError as ex:
 90            if (raise_on_404 or (ex.response is None) or (ex.response.status_code != http.HTTPStatus.NOT_FOUND)):
 91                raise ex
 92
 93            return None
 94
 95        next_url = fetch_next_canvas_link(response)
 96
 97        new_results = edq.util.json.loads(body_text, strict = True)
 98        for new_result in new_results:
 99            output.append(new_result)
100
101    return output
102
103def parse_timestamp(value: typing.Union[str, None]) -> typing.Union[edq.util.time.Timestamp, None]:
104    """ Parse a Canvas-style timestamp into a common form. """
105
106    if (value is None):
107        return None
108
109    # Parse out some cases that Python <= 3.10 cannot deal with.
110    value = re.sub(r'Z$', '+00:00', value)
111    value = re.sub(r'(\d\d:\d\d)(\.\d+)', r'\1', value)
112
113    pytime = datetime.datetime.fromisoformat(value)
114    return edq.util.time.Timestamp.from_pytime(pytime)