Module pyaurorax.search.api.classes.request

Class definition for an AuroraX API request

Classes

class AuroraXAPIRequest (aurorax_obj,
url: str,
method: Literal['get', 'post', 'put', 'delete', 'patch'],
params: Dict = {},
body: List | Dict = {},
headers: Dict = {},
null_response: bool = False)

Expand source code

class AuroraXAPIRequest:
    """
    Class definition for an AuroraX API request

    Attributes:
        url (str): 
            API endpoint URL for the request

        method (str): 
            The HTTP method to use. Valid values are: `get`, `post`, `put`, `delete`, `patch`

        params (Dict): 
            URL parameters to send in the request, defaults to `{}`

        body (Dict): 
            The body of the request (ie. post data), defaults to `{}`

        headers (Dict): 
            Any headers to send as part of the request (in addition to the default ones), defaults to `{}`
            
        null_response (bool): 
            Signifies if we expect a response from the API that has no body/data in it (ie. 
            requests to upload data that respond with just a 202 status code), defaults to `False`
    """

    __API_KEY_HEADER_NAME = "x-aurorax-api-key"

    def __init__(self,
                 aurorax_obj,
                 url: str,
                 method: Literal["get", "post", "put", "delete", "patch"],
                 params: Dict = {},
                 body: Union[List, Dict] = {},
                 headers: Dict = {},
                 null_response: bool = False):
        self.__aurorax_obj = aurorax_obj
        self.url = url
        self.method = method
        self.params = params
        self.body = body
        self.headers = headers
        self.null_response = null_response

    def __json_converter(self, o):
        # NOTE: this method is a last-ditch catch for any datetimes that snuck
        # into the request body without already being converted to string format.
        # Since this should never happen, we exclude it from the test suite
        if (isinstance(o, datetime.datetime) is True):  # pragma: nocover-ok
            return str(o)

    def __merge_headers(self):
        # set initial headers
        all_headers = self.__aurorax_obj.api_headers

        # add headers passed into the class
        #
        # NOTE: we don't usually pass any extra headers, so exclude this from
        # the test suite
        for key, value in self.headers.items():  # pragma: nocover-ok
            all_headers[key.lower()] = value

        # add api key
        if self.__aurorax_obj.api_key:
            all_headers[self.__API_KEY_HEADER_NAME] = self.__aurorax_obj.api_key

        # return
        return all_headers

    def execute(self) -> AuroraXAPIResponse:
        """
        Execute an AuroraX API request
    
        Returns:
            An `pyaurorax.search.api.AuroraXAPIResponse` object

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: Error during API call
            pyaurorax.exceptions.AuroraXUnauthorizedError: Unauthorized user attempted to perform this action
            pyaurorax.exceptions.AuroraXMaintenanceError: AuroraX API is currently undergoing maintenance
        """
        # sanitize data
        body_santized = json.dumps(self.body, default=self.__json_converter)

        # make request
        try:
            req = requests.request(self.method,
                                   self.url,
                                   headers=self.__merge_headers(),
                                   params=self.params,
                                   data=body_santized,
                                   timeout=self.__aurorax_obj.api_timeout)
        except requests.exceptions.Timeout:  # pragma: nocover-ok
            raise AuroraXAPIError("API request timeout reached") from None

        # check if authorization worked (raised by API or Nginx)
        if (req.status_code == 401):  # pragma: nocover-ok
            if (req.headers["Content-Type"] == "application/json"):
                if ("error_message" in req.json()):
                    # this will be an error message that the API meant to send
                    raise AuroraXUnauthorizedError("API error code %d: %s" % (req.status_code, req.json()["error_message"]))
                else:
                    raise AuroraXUnauthorizedError("API error code 401: unauthorized")
            else:
                raise AuroraXUnauthorizedError("API error code 401: unauthorized")

        # check for 404 error (raised by API or by Nginx)
        if (req.status_code == 404):  # pragma: nocover-ok
            if (req.headers["Content-Type"] == "application/json"):
                if ("error_message" in req.json()):
                    # this will be an error message that the API meant to send
                    raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.json()["error_message"]))
                else:
                    # this will likely be a 404 from the API
                    raise AuroraXAPIError("API error code 404: not found")
            else:
                raise AuroraXAPIError("API error code 404: not found")

        # check for 400
        if (req.status_code == 400):  # pragma: nocover-ok
            raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))

        # check for server error
        if (req.status_code == 500):  # pragma: nocover-ok
            try:
                response_json = req.json()
                if ("error_message" in response_json):
                    raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json["error_message"]))
                else:
                    raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json))
            except Exception as e:
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content)) from e

        # check for maintenance mode error
        if (req.status_code == 502):  # pragma: nocover-ok
            raise AuroraXAPIError("API error code %d: API inaccessible, bad gateway" % (req.status_code))

        # check for maintenance mode error
        if (req.status_code == 503):  # pragma: nocover-ok
            response_json = req.json()
            if ("maintenance mode" in response_json["error_message"].lower()):
                raise AuroraXMaintenanceError(response_json["error_message"])
            else:
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json["error_message"]))

        # check content type
        if (self.null_response is False):  # pragma: nocover-ok
            if (req.headers["Content-Type"] == "application/json"):
                if (len(req.content) == 0):
                    raise AuroraXAPIError("API error code %d: no response received" % (req.status_code))
                else:
                    response_data = req.json()
            else:
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))
        else:
            if (req.status_code in [200, 201, 202, 204]):
                response_data = None
            else:
                response_data = req.json()

        # create response object
        res = AuroraXAPIResponse(request=req, data=response_data, status_code=req.status_code)

        # return
        return res

    def __str__(self) -> str:
        return self.__repr__()  # pragma: nocover-ok

    def __repr__(self) -> str:
        return f"AuroraXAPIRequest(method='{self.method}', url='{self.url}')"  # pragma: nocover-ok

Class definition for an AuroraX API request

Attributes

-----=

url : str

API endpoint URL for the request

method : str

The HTTP method to use. Valid values are: get, post, put, delete, patch

params : Dict

URL parameters to send in the request, defaults to {}

body : Dict

The body of the request (ie. post data), defaults to {}

headers : Dict

Any headers to send as part of the request (in addition to the default ones), defaults to {}

null_response : bool

Signifies if we expect a response from the API that has no body/data in it (ie. requests to upload data that respond with just a 202 status code), defaults to False

Methods

def execute(self) ‑> AuroraXAPIResponse

Expand source code

def execute(self) -> AuroraXAPIResponse:
    """
    Execute an AuroraX API request

    Returns:
        An `pyaurorax.search.api.AuroraXAPIResponse` object

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: Error during API call
        pyaurorax.exceptions.AuroraXUnauthorizedError: Unauthorized user attempted to perform this action
        pyaurorax.exceptions.AuroraXMaintenanceError: AuroraX API is currently undergoing maintenance
    """
    # sanitize data
    body_santized = json.dumps(self.body, default=self.__json_converter)

    # make request
    try:
        req = requests.request(self.method,
                               self.url,
                               headers=self.__merge_headers(),
                               params=self.params,
                               data=body_santized,
                               timeout=self.__aurorax_obj.api_timeout)
    except requests.exceptions.Timeout:  # pragma: nocover-ok
        raise AuroraXAPIError("API request timeout reached") from None

    # check if authorization worked (raised by API or Nginx)
    if (req.status_code == 401):  # pragma: nocover-ok
        if (req.headers["Content-Type"] == "application/json"):
            if ("error_message" in req.json()):
                # this will be an error message that the API meant to send
                raise AuroraXUnauthorizedError("API error code %d: %s" % (req.status_code, req.json()["error_message"]))
            else:
                raise AuroraXUnauthorizedError("API error code 401: unauthorized")
        else:
            raise AuroraXUnauthorizedError("API error code 401: unauthorized")

    # check for 404 error (raised by API or by Nginx)
    if (req.status_code == 404):  # pragma: nocover-ok
        if (req.headers["Content-Type"] == "application/json"):
            if ("error_message" in req.json()):
                # this will be an error message that the API meant to send
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.json()["error_message"]))
            else:
                # this will likely be a 404 from the API
                raise AuroraXAPIError("API error code 404: not found")
        else:
            raise AuroraXAPIError("API error code 404: not found")

    # check for 400
    if (req.status_code == 400):  # pragma: nocover-ok
        raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))

    # check for server error
    if (req.status_code == 500):  # pragma: nocover-ok
        try:
            response_json = req.json()
            if ("error_message" in response_json):
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json["error_message"]))
            else:
                raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json))
        except Exception as e:
            raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content)) from e

    # check for maintenance mode error
    if (req.status_code == 502):  # pragma: nocover-ok
        raise AuroraXAPIError("API error code %d: API inaccessible, bad gateway" % (req.status_code))

    # check for maintenance mode error
    if (req.status_code == 503):  # pragma: nocover-ok
        response_json = req.json()
        if ("maintenance mode" in response_json["error_message"].lower()):
            raise AuroraXMaintenanceError(response_json["error_message"])
        else:
            raise AuroraXAPIError("API error code %d: %s" % (req.status_code, response_json["error_message"]))

    # check content type
    if (self.null_response is False):  # pragma: nocover-ok
        if (req.headers["Content-Type"] == "application/json"):
            if (len(req.content) == 0):
                raise AuroraXAPIError("API error code %d: no response received" % (req.status_code))
            else:
                response_data = req.json()
        else:
            raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))
    else:
        if (req.status_code in [200, 201, 202, 204]):
            response_data = None
        else:
            response_data = req.json()

    # create response object
    res = AuroraXAPIResponse(request=req, data=response_data, status_code=req.status_code)

    # return
    return res

Execute an AuroraX API request

Returns -----= An AuroraXAPIResponse object

Raises

-----=

AuroraXAPIError

Error during API call

AuroraXUnauthorizedError

Unauthorized user attempted to perform this action

AuroraXMaintenanceError

AuroraX API is currently undergoing maintenance