121 lines
6 KiB
Python
121 lines
6 KiB
Python
import logging
|
|
from typing import Dict
|
|
from json import JSONDecodeError
|
|
|
|
import requests
|
|
from urllib3 import disable_warnings
|
|
|
|
from pyzipline.errors import KwargConflict, HTTPFailure, PyZiplineError
|
|
from pyzipline.models import Result
|
|
|
|
class RestAdapter:
|
|
def __init__(self, hostname: str, token: str = '', ssl: bool = True, enforced_signing: bool = True, logger: logging.Logger = None):
|
|
"""Constructor for RestAdapter
|
|
|
|
:param hostname: The hostname of your Zipline instance, WITHOUT https or http.
|
|
:type hostname: str
|
|
:param token: (optional) String used for authentication when making requests.
|
|
:param token: str
|
|
:param ssl: (optional) Normally set to True, but if your Zipline instance doesn't use SSL/TLS, set this to False.
|
|
:type ssl: bool
|
|
:param enforced_signing: (optional) Normally set to True, but if having SSL/TLS cert validation issues, can turn off with False.
|
|
:type enforced_signing: bool
|
|
:param logger: (optional) If your app has a logger, pass it in here.
|
|
:type logger: logging.Logger
|
|
:raise KwargConflict: Raised when the keyword arguments passed to a function conflict.
|
|
"""
|
|
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 KwargConflict("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, data: Dict = None) -> Result:
|
|
"""Internal method to make a request to the Zipline server. You shouldn't use this directly.
|
|
|
|
:param http_method: The HTTP method to use (GET, POST, DELETE)
|
|
:type http_method: str
|
|
:param endpoint: The endpoint to make the request to.
|
|
:type endpoint: str
|
|
:param params: (optional) Python dictionary of query parameters to send with the request.
|
|
:type params: Dict
|
|
:param data: (optional) Python dictionary of data to send with the request.
|
|
:type data: Dict
|
|
:raise HTTPFailure: Raised when an HTTP request fails.
|
|
:raise PyZiplineError: Raised when an error occurs in the PyZipline library.
|
|
:return: Result object
|
|
:rtype: Result"""
|
|
full_url = self._url + endpoint
|
|
headers = {'Authorization': self._token}
|
|
|
|
log_line_pre = f"method={http_method}, url={full_url}, params={params}"
|
|
log_line_post = ', '.join((log_line_pre, "success={}, status_code={}, message={}"))
|
|
|
|
try: # Log HTTP params and perform an HTTP request, catching and re-raising any exceptions
|
|
self._logger.debug(msg=log_line_pre)
|
|
# 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=data)
|
|
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:
|
|
self._logger.error(msg=log_line_post.format(False, None, 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 raise exception
|
|
is_success = 299 >= response.status_code >= 200
|
|
log_line = log_line_post.format(is_success, response.status_code, response.reason)
|
|
|
|
if is_success:
|
|
self._logger.debug(msg=log_line_post.format(is_success, response.status_code, response.reason))
|
|
return Result(status_code=response.status_code, message=response.reason, data=data_out)
|
|
|
|
self._logger.error(msg=log_line)
|
|
|
|
raise PyZiplineError(f"{response.status_code}: {response.reason}")
|
|
|
|
def get(self, endpoint: str, params: Dict = None) -> Result:
|
|
"""Make a GET request to the Zipline server. You should almost never have to use this directly.
|
|
|
|
:param endpoint: The endpoint to make the request to.
|
|
:type endpoint: str
|
|
:param params: (optional) Python dictionary of query parameters to send with the request.
|
|
:type params: Dict
|
|
:return: Result object
|
|
:rtype: Result"""
|
|
return self._do(http_method='GET', endpoint=endpoint, params=params)
|
|
|
|
def post(self, endpoint: str, params: Dict = None, data: Dict = None) -> Result:
|
|
"""Make a POST request to the Zipline server. You should almost never have to use this directly.
|
|
|
|
:param endpoint: The endpoint to make the request to.
|
|
:type endpoint: str
|
|
:param params: (optional) Python dictionary of query parameters to send with the request.
|
|
:type params: Dict
|
|
:param data: (optional) Python dictionary of data to send with the request.
|
|
:type data: Dict
|
|
:return: Result object
|
|
:rtype: Result"""
|
|
return self._do(http_method='POST', endpoint=endpoint, params=params, data=data)
|
|
|
|
def delete(self, endpoint: str, params: Dict = None, data: Dict = None) -> Result:
|
|
"""Make a DELETE request to the Zipline server. You should almost never have to use this directly.
|
|
|
|
:param endpoint: The endpoint to make the request to.
|
|
:type endpoint: str
|
|
:param params: (optional) Python dictionary of query parameters to send with the request.
|
|
:type params: Dict
|
|
:param data: (optional) Python dictionary of data to send with the request.
|
|
:type data: Dict
|
|
:return: Result object
|
|
:rtype: Result"""
|
|
return self._do(http_method='DELETE', endpoint=endpoint, params=params, data=data)
|