autograder.api.common

  1import os
  2import sys
  3import typing
  4
  5import edq.core.errors
  6import edq.net.request
  7import edq.util.json
  8import edq.util.time
  9import requests
 10
 11import autograder
 12import autograder.api.config
 13import autograder.api.constants
 14import autograder.error
 15import autograder.model.config
 16
 17DEFAULT_SOURCE_NAME: str = 'edq-autograder-py'
 18DEFAULT_SOURCE_VERSION: str = autograder.__version__
 19
 20TESTING_SOURCE_NAME: str = 'testing'
 21TESTING_SOURCE_VERSION: str = '0.0.0'
 22
 23_source_name: str = DEFAULT_SOURCE_NAME  # pylint: disable=invalid-name
 24_source_version: str = DEFAULT_SOURCE_VERSION  # pylint: disable=invalid-name
 25
 26def set_testing_source_info() -> None:
 27    """ Set source info for API requests to consistent values for testing. """
 28
 29    global _source_name  # pylint: disable=global-statement
 30    global _source_version  # pylint: disable=global-statement
 31
 32    _source_name = TESTING_SOURCE_NAME
 33    _source_version = TESTING_SOURCE_VERSION
 34
 35def make_api_request(
 36        endpoint: str,
 37        config: autograder.model.config.Config,
 38        api_params: typing.List[autograder.api.config.APIParam],
 39        write: typing.Union[bool, None] = None,
 40        exit_on_error: bool = False,
 41        post_paths: typing.Union[typing.List[str], None] = None,
 42        ) -> typing.Dict[str, typing.Any]:
 43    """
 44    Given arguments (usually from argparse), API params, and an endpoint,
 45    make an API request.
 46    Will raise an error on any error or error response.
 47
 48    Requests that MAY write data to the server should pass `write` as true.
 49    When set, a header (autograder.api.constants.HEADER_KEY_WRITE) will be sent along with the API request.
 50    This should only affect testing, and will allow server runners to restart when write operations are sent.
 51    """
 52
 53    try:
 54        return _make_api_request(endpoint, config, api_params, post_paths = post_paths, write = write)
 55    except autograder.error.AutograderError as ex:
 56        if (exit_on_error):
 57            print("ERROR: " + ex.args[0], file = sys.stderr)
 58            autograder.error.exit_from_error(1)
 59
 60        raise ex
 61
 62def _make_api_request(
 63        endpoint: str,
 64        config: autograder.model.config.Config,
 65        api_params: typing.List[autograder.api.config.APIParam],
 66        write: typing.Union[bool, None] = None,
 67        post_paths: typing.Union[typing.List[str], None] = None,
 68        ) -> typing.Dict[str, typing.Any]:
 69    """ Wrapped function for make_api_request(). """
 70
 71    payload = _verify_payload(config, api_params)
 72
 73    return send_api_request(endpoint, payload, post_paths = post_paths, write = write)
 74
 75def _verify_payload(
 76        config: autograder.model.config.Config,
 77        api_params: typing.List[autograder.api.config.APIParam],
 78        ) -> typing.Dict[str, typing.Any]:
 79    """
 80    Verify that the given payload matches the listed API parameters,
 81    and return a new copy of the payload with only the specified keys.
 82    """
 83
 84    payload = {}
 85
 86    for api_param in api_params:
 87        if (not api_param.api):
 88            continue
 89
 90        value = api_param.clean_value(getattr(config, api_param.config_key, None))
 91
 92        # If there is no value, try an alt key.
 93        if ((value is None) and (api_param.alt_config_key is not None)):
 94            value = api_param.clean_value(getattr(config, api_param.alt_config_key, None))
 95
 96        if (value is None):
 97            if (api_param.api_required):
 98                names = [f"'{name}'" for name in [api_param.config_key, api_param.alt_config_key, api_param.cli_flag] if (name is not None)]
 99                names = sorted(set(names))
100
101                raise autograder.error.APIError(None,
102                    f"Required parameter {' / '.join(names)} not found.")
103
104            if (api_param.omit_empty):
105                continue
106
107        payload[api_param.api_key] = value
108
109    return payload
110
111def send_api_request(
112        endpoint: str,
113        payload: typing.Dict[str, typing.Any],
114        write: typing.Union[bool, None] = None,
115        post_paths: typing.Union[typing.List[str], None] = None,
116        ) -> typing.Dict[str, typing.Any]:
117    """
118    Make an autograder API request.
119    On a failure (including non 200), an error will be raised with error information.
120    Otherwise (including a successful response with a failed operation (soft error)),
121    the content of the response will be returned (converted from JSON to a dict).
122    """
123
124    if (post_paths is None):
125        post_paths = []
126
127    server = payload.pop('server', None)
128    if (server is None):
129        raise autograder.error.APIError(None, "No server provided.")
130
131    server = server.rstrip('/')
132    endpoint = endpoint.lstrip('/')
133
134    url = f"{server}/api/{autograder.api.constants.API_VERSION}/{endpoint}"
135
136    # Add source information.
137    payload['source'] = _source_name
138    payload['source-version'] = _source_version
139
140    headers = {}
141    if (write is not None):
142        headers[autograder.api.constants.HEADER_KEY_WRITE] = str(write).lower()
143
144    post_files = {}
145    for path in post_paths:
146        filename = os.path.basename(path)
147        if (filename in post_files):
148            raise autograder.error.APIError(None, f"Cannot submit duplicate filenames ('{filename}').")
149
150        if (not os.path.isfile(path)):
151            raise autograder.error.APIError(None, f"Path does not exist, or is not a file: '{path}'.")
152
153        post_files[filename] = open(path, 'rb')  # pylint: disable=consider-using-with
154
155    try:
156        raw_response, raw_body = edq.net.request.make_post(
157            url,
158            data = {autograder.api.constants.API_REQUEST_JSON_KEY: edq.util.json.dumps(payload)},
159            headers = headers,
160            files = post_files,
161            raise_for_status = False)
162    except edq.core.errors.RetryError as ex:
163        # If this is not a connection error, then re-raise.
164        if (not ex.contains_instance(requests.exceptions.ConnectionError)):
165            raise ex
166
167        raise autograder.error.ConnectionError((f"Could not connect to autograder server '{server}'."  # pylint: disable=raise-missing-from
168            + " This is a networking issue (e.g., network down, server down, wrong server address), not an authentication issue."))
169
170    for file in post_files.values():
171        file.close()
172
173    try:
174        response_body = edq.util.json.loads(raw_body, strict = True)
175    except Exception as ex:
176        message = ("Autograder response does not contain valid JSON."
177            + " Contact a server admin with the following."
178            + f" Response:\n---\n{raw_response.text}\n---")
179
180        raise autograder.error.APIError(None, message) from ex
181
182    if (not response_body.get(autograder.api.constants.API_RESPONSE_KEY_SUCCESS, False)):
183        message = 'Request to the autograder failed.'
184        if (autograder.api.constants.API_RESPONSE_KEY_MESSAGE in response_body):
185            message = f"Failed to complete operation: {response_body[autograder.api.constants.API_RESPONSE_KEY_MESSAGE]}"
186
187            # Replace any timestamps in the message.
188            message = edq.util.time.Timestamp.convert_embedded(message, pretty = True)
189
190        code = response_body.get(autograder.api.constants.API_RESPONSE_KEY_STATUS, None)
191        raise autograder.error.APIError(code, message)
192
193    content = response_body[autograder.api.constants.API_RESPONSE_KEY_CONTENT]
194    return typing.cast(typing.Dict[str, typing.Any], content)