Module pyaurorax.search.api

Interface for AuroraX API requests. Primarily an under-the-hood module not needed for most use-cases.

Sub-modules

pyaurorax.search.api.classes

Class definitions used by the api submodule

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):
        if (isinstance(o, datetime.datetime) is True):
            return str(o)

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

        # add headers passed into the class
        for key, value in self.headers.items():
            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:
            raise AuroraXAPIError("API request timeout reached") from None

        # check if authorization worked (raised by API or Nginx)
        if (req.status_code == 401):
            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):
            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):
            raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))

        # check for server error
        if (req.status_code == 500):
            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))

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

        # check for maintenance mode error
        if (req.status_code == 503):
            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):
            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__()

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

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:
        raise AuroraXAPIError("API request timeout reached") from None

    # check if authorization worked (raised by API or Nginx)
    if (req.status_code == 401):
        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):
        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):
        raise AuroraXAPIError("API error code %d: %s" % (req.status_code, req.content.decode()))

    # check for server error
    if (req.status_code == 500):
        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))

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

    # check for maintenance mode error
    if (req.status_code == 503):
        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):
        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

class AuroraXAPIResponse (request: Any, data: Any, status_code: int)

Expand source code

class AuroraXAPIResponse:
    """
    Class definition for an AuroraX API response

    Attributes:
        request (Any): 
            The request object

        data (Any): 
            The data received as part of the request

        status_code (int): 
            The HTTP status code received when making the request
    """

    def __init__(self, request: Any, data: Any, status_code: int):
        self.request = request
        self.data = data
        self.status_code = status_code

    def __str__(self) -> str:
        return self.__repr__()

    def __repr__(self) -> str:
        return f"AuroraXAPIResponse [{self.status_code}] ({responses[self.status_code]})"

Class definition for an AuroraX API response

Attributes

request : Any

The request object

data : Any

The data received as part of the request

status_code : int

The HTTP status code received when making the request