PyZipline/pyzipline/rest_adapter.py

"""This module contains the RestAdapter class, which is used to make requests to the Zipline server."""""
import logging
from typing import Dict
from json import JSONDecodeError

import requests
from urllib3 import disable_warnings

from pyzipline.exceptions import HTTPFailure, PyZiplineError
from pyzipline.models import Result

class RestAdapter:
    """Constructor for RestAdapter

    Args:
        hostname (str): The hostname of your Zipline instance, WITHOUT https or http.
        token (str = None): String used for authentication when making requests.
        ssl (bool = True): Normally set to True, but if your Zipline instance doesn't use SSL/TLS, set this to False.
        enforced_signing (bool = True): Normally set to True, but if having SSL/TLS cert validation issues, can turn off with False.
        logger (logging.Logger = None): If your app has a logger, pass it in here.

    Raises:
        ValueError: Raised when the keyword arguments passed to the class constructor conflict.
    """
    def __init__(self, hostname: str, token: str = '', ssl: bool = True, enforced_signing: bool = True, logger: logging.Logger = None):
        self._url = f"http{'s' if ssl else ''}://{hostname}/api/"
        self._token = token
        self._ssl = ssl
        self._enforced_signing = enforced_signing
        self._logger = logger or logging.getLogger(__name__)

        if ssl is False and enforced_signing is True:
            raise ValueError("Cannot enforce signing without SSL")

        if not ssl and not enforced_signing:
            disable_warnings()

    def _do(self, http_method: str, endpoint: str, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result:
        """Internal method to make a request to the Zipline server. You shouldn't use this directly.

        Args:
            http_method (str): The [HTTP method](https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods) to use
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            json (dict): Python dictionary of JSON-serializable data to send with the request.
            files (dict): Python dictionary of files to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Returns:
            Result: A Result object containing the status code, message, and data from the request.

        Raises:
            HTTPError: Raised when an HTTP request fails.
            PyZiplineError: Raised when an error occurs in the PyZipline library.
        """
        full_url = self._url + endpoint
        headers = {'Authorization': self._token}

        try: # Perform an HTTP request, catching and re-raising any exceptions
            # will eventually refactor this to use asyncio/aiohttp instead for async operation
            response = requests.request(method=http_method, url=full_url, verify=self._enforced_signing, params=params, headers=headers, json=json, files=files, timeout=timeout)
        except requests.exceptions.RequestException as e:
            self._logger.error(msg=str(e))
            raise HTTPFailure("Could not connect to Zipline server") from e

        try: # Deserialize JSON output to Python object, or return failed Result on exception
            data_out = response.json()
        except (ValueError, JSONDecodeError) as e:
            raise PyZiplineError("Could not decode response from Zipline server") from e

        # If status_code in 200-299 range, return success Result with data, otherwise return failed Result with message
        is_success = 299 >= response.status_code >= 200

        if is_success:
            return Result(success=is_success, status_code=response.status_code, message=response.reason, data=data_out)

        return Result(success=is_success, status_code=response.status_code, message=data_out['error'], data=data_out)

    def delete(self, endpoint: str, params: dict = None, json: dict = None, timeout: float = 60) -> Result:
        """Make a DELETE request to the Zipline server. You should almost never have to use this directly.

        Args:
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            json (dict): Python dictionary of JSON-serializable data to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        return self._do(http_method='DELETE', endpoint=endpoint, params=params, json=json, timeout=timeout)

    def get(self, endpoint: str, params: dict = None, timeout: float = 60) -> Result:
        """Make a GET request to the Zipline server. You should almost never have to use this directly.

        Args:
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        return self._do(http_method='GET', endpoint=endpoint, params=params, timeout=timeout)

    def patch(self, endpoint: str, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result:
        """Make a PATCH request to the Zipline server. You should almost never have to use this directly.

        Args:
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            json (dict): Python dictionary of JSON-serializable data to send with the request.
            files (dict): Python dictionary of files to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        return self._do(http_method='PATCH', endpoint=endpoint, params=params, json=json, files=files, timeout=timeout)

    def post(self, endpoint: str, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result:
        """Make a POST request to the Zipline server. You should almost never have to use this directly.

        Args:
            endpoint (str): The endpoint to make the request to.
            params (dict): Python dictionary of query parameters to send with the request.
            json (dict): Python dictionary of JSON-serializable data to send with the request.
            files (dict): Python dictionary of files to send with the request.
            timeout (float): Number of seconds to wait for the request to complete.

        Returns:
            Result: A Result object containing the status code, message, and data from the request.
        """
        return self._do(http_method='POST', endpoint=endpoint, params=params, json=json, files=files, timeout=timeout)