cswimr/PyZipline
cswimr/PyZipline
1
1
Fork 0
Code Issues Pull requests Releases Packages Wiki Activity Actions

misc: various rest_adapter changes

Browse source
This commit is contained in:
Seaswimmer 2023-12-22 13:48:21 -05:00
parent e47b4de047
commit 9794552632
Signed by: cswimr
GPG key ID: 1EBC234EEDA901AE
1 changed files with 55 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
pyzipline/rest_adapter.py
View file

@ -35,15 +35,16 @@ class RestAdapter:
if not ssl and not enforced_signing:
disable_warnings()
def _do(self, http_method: str, endpoint: str, params: Dict = None, data: Dict = None, timeout: float = 60) -> Result:
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 to use (GET, POST, DELETE)
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 = None): Python dictionary of query parameters to send with the request.
data (Dict = None): Python dictionary of data to send with the request.
timeout (float = 60): Number of seconds to wait for the request to complete.
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.
@ -57,7 +58,7 @@ class RestAdapter:
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=data, timeout=timeout)
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
@ -75,40 +76,59 @@ class RestAdapter:
return Result(success=is_success, status_code=response.status_code, message=data_out['error'], data=data_out)
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.
Args:
endpoint (str): The endpoint to make the request to.
params (Dict = None): Python dictionary of query parameters to send with the request.
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)
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.
Args:
endpoint (str): The endpoint to make the request to.
params (Dict = None): Python dictionary of query parameters to send with the request.
data (Dict = None): Python dictionary of data to send with the request.
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, data=data)
def delete(self, endpoint: str, params: Dict = None, data: Dict = None) -> Result:
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 = None): Python dictionary of query parameters to send with the request.
data (Dict = None): Python dictionary of data 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.
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, data=data)
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)