Skip to content

Rest Adapter

Internal Functions

These functions are meant for use in other parts of the module. You probably shouldn't be calling these manually.
If there's an endpoint method missing from the main ZiplineApi class, you should open an issue (or a pull request).

This module contains the RestAdapter class, which is used to make requests to the Zipline server.

RestAdapter

Constructor for RestAdapter

Parameters:

Name Type Description Default
config ZiplineApiConfig

Configuration object for the ZiplineApi class

 ZiplineApiConfig

Raises:

Type Description
ValueError

Raised when the keyword arguments passed to the class constructor conflict.
Source code in pyzipline/rest_adapter.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class RestAdapter:
    """Constructor for RestAdapter

    Args:
        config (ZiplineApiConfig): Configuration object for the ZiplineApi class

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

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

        if not config.ssl and not config.enforced_signing:
            disable_warnings()

    def _do(self, http_method: str, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
        """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.
            headers (dict): Python dictionary of headers to send with the request.
            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
        if not headers:
            headers = {}
        headers.update({'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:
            if response.headers.get('Content-Type') == 'application/octet-stream':
                # If the response contains a valid file, return success Result with file content
                return Result(success=True, status_code=response.status_code, message=response.reason, data=response.content)
            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, headers: dict = None, params: dict = None, json: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
        """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, headers=headers, params=params, json=json, timeout=timeout)

    def get(self, endpoint: str, headers: dict = None, 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, headers=headers, params=params, timeout=timeout)

    def patch(self, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
        """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, headers=headers, params=params, json=json, files=files, timeout=timeout)

    def post(self, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
        """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.
            headers (dict): Python dictionary of headers to send with the request.
            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, headers=headers, params=params, json=json, files=files, timeout=timeout)

_do(http_method, endpoint, headers=None, params=None, json=None, files=None, timeout=60)

Internal method to make a request to the Zipline server. You shouldn't use this directly.

Parameters:

Name Type Description Default
http_method str

The HTTP method to use

 required
endpoint str

The endpoint to make the request to.

 required
headers dict

Python dictionary of headers to send with the request.

 None
params dict

Python dictionary of query parameters to send with the request.

 None
json dict

Python dictionary of JSON-serializable data to send with the request.

 None
files dict

Python dictionary of files to send with the request.

 None
timeout float

Number of seconds to wait for the request to complete.

 60

Returns:

Name Type Description
Result Result

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

Raises:

Type Description
HTTPError

Raised when an HTTP request fails.
PyZiplineError

Raised when an error occurs in the PyZipline library.
Source code in pyzipline/rest_adapter.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def _do(self, http_method: str, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
    """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.
        headers (dict): Python dictionary of headers to send with the request.
        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
    if not headers:
        headers = {}
    headers.update({'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:
        if response.headers.get('Content-Type') == 'application/octet-stream':
            # If the response contains a valid file, return success Result with file content
            return Result(success=True, status_code=response.status_code, message=response.reason, data=response.content)
        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)

delete(endpoint, headers=None, params=None, json=None, timeout=60)

Make a DELETE request to the Zipline server. You should almost never have to use this directly.

Parameters:

Name Type Description Default
endpoint str

The endpoint to make the request to.

 required
params dict

Python dictionary of query parameters to send with the request.

 None
json dict

Python dictionary of JSON-serializable data to send with the request.

 None
timeout float

Number of seconds to wait for the request to complete.

 60

Returns:

Name Type Description
Result Result

A Result object containing the status code, message, and data from the request.
Source code in pyzipline/rest_adapter.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
def delete(self, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
    """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, headers=headers, params=params, json=json, timeout=timeout)

get(endpoint, headers=None, params=None, timeout=60)

Make a GET request to the Zipline server. You should almost never have to use this directly.

Parameters:

Name Type Description Default
endpoint str

The endpoint to make the request to.

 required
params dict

Python dictionary of query parameters to send with the request.

 None
timeout float

Number of seconds to wait for the request to complete.

 60

Returns:

Name Type Description
Result Result

A Result object containing the status code, message, and data from the request.
Source code in pyzipline/rest_adapter.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def get(self, endpoint: str, headers: dict = None, 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, headers=headers, params=params, timeout=timeout)

patch(endpoint, headers=None, params=None, json=None, files=None, timeout=60)

Make a PATCH request to the Zipline server. You should almost never have to use this directly.

Parameters:

Name Type Description Default
endpoint str

The endpoint to make the request to.

 required
params dict

Python dictionary of query parameters to send with the request.

 None
json dict

Python dictionary of JSON-serializable data to send with the request.

 None
files dict

Python dictionary of files to send with the request.

 None
timeout float

Number of seconds to wait for the request to complete.

 60

Returns:

Name Type Description
Result Result

A Result object containing the status code, message, and data from the request.
Source code in pyzipline/rest_adapter.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def patch(self, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
    """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, headers=headers, params=params, json=json, files=files, timeout=timeout)

post(endpoint, headers=None, params=None, json=None, files=None, timeout=60)

Make a POST request to the Zipline server. You should almost never have to use this directly.

Parameters:

Name Type Description Default
endpoint str

The endpoint to make the request to.

 required
headers dict

Python dictionary of headers to send with the request.

 None
params dict

Python dictionary of query parameters to send with the request.

 None
json dict

Python dictionary of JSON-serializable data to send with the request.

 None
files dict

Python dictionary of files to send with the request.

 None
timeout float

Number of seconds to wait for the request to complete.

 60

Returns:

Name Type Description
Result Result

A Result object containing the status code, message, and data from the request.
Source code in pyzipline/rest_adapter.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def post(self, endpoint: str, headers: dict = None, params: dict = None, json: dict = None, files: dict = None, timeout: float = 60) -> Result: # pylint: disable=too-many-arguments
    """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.
        headers (dict): Python dictionary of headers to send with the request.
        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, headers=headers, params=params, json=json, files=files, timeout=timeout)