34 lines
1.4 KiB
Python
34 lines
1.4 KiB
Python
|
from typing import Dict, List
|
||
|
|
|
||
|
|
import requests
|
||
|
|
from urllib3 import disable_warnings
|
||
|
|
|
||
|
|
from .errors import KwargConflict
|
||
|
|
|
||
|
|
class RestAdapter:
|
||
|
|
def __init__(self, hostname: str, api_key: str = '', ssl: bool = True, enforced_signing: bool = True):
|
||
|
|
"""Create a new RestAdapter instance, to interact with the REST API of a Zipline server.
|
||
|
|
|
||
|
|
:param hostname: The hostname of the Zipline server
|
||
|
|
:param api_key: The API key to use for authentication
|
||
|
|
:param ssl: Whether to use SSL
|
||
|
|
:param enforced_signing: Whether to enforce SSL certificate signing
|
||
|
|
"""
|
||
|
|
self.url = f"http{'s' if ssl else ''}://{hostname}/"
|
||
|
|
self.api_key = api_key
|
||
|
|
self.ssl = ssl
|
||
|
|
self.enforced_signing = enforced_signing
|
||
|
|
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 get(self, endpoint: str) -> List[Dict]:
|
||
|
|
full_url = self.url + endpoint
|
||
|
|
headers = {'Authorization': self.api_key}
|
||
|
|
response = requests.get(url=full_url, verify=self.enforced_signing, headers=headers)
|
||
|
|
data_out = response.json()
|
||
|
|
if response.status_code >= 200 and response.status_code <= 299: # OK
|
||
|
|
return data_out
|
||
|
|
raise Exception(data_out["message"]) # Todo: raise custom exception later
|