Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_paasta_api_client():
return SwaggerClient.from_url(get_paasta_api_url())
to_wire=lambda uri: uri,
to_python=lambda uri: uri,
validate=lambda uri_string: True
)
bravado_config = {
'also_return_response': True,
'formats': [uri_format]
}
service = None
start_time = time.perf_counter()
cur_time = start_time
while cur_time < start_time + 10:
try:
service = SwaggerClient.from_url('http://localhost:29593/swagger.json', config=bravado_config)
_, response = service.jobs.get_jobs().result()
if response.status_code == 200:
break
except HTTPBadGateway:
pass
except requests.exceptions.ConnectionError:
pass
time.sleep(0.1)
cur_time = time.perf_counter()
if cur_time >= start_time + 10:
print("Warning: Cerise container failed to come up")
return service
def construct_swagger_clients(hostname, token):
http_client = RequestsClient()
http_client.session.verify = False
http_client.session.headers = {
'Authorization': 'Bearer %s' % token
}
specs_client = SwaggerClient.from_url(hostname + SWAGGER_SPECS_PREFIX, http_client=http_client,
config={'validate_responses': False, 'also_return_response': False})
docs_client = SwaggerDocsClient.from_url(hostname + SWAGGER_DOCS_PREFIX, http_client=http_client)
return specs_client, docs_client
import pprint
HOST = "https://testnet.bitmex.com"
SPEC_URI = HOST + "/api/explorer/swagger.json"
# See full config options at http://bravado.readthedocs.io/en/latest/configuration.html
config = {
# Don't use models (Python classes) instead of dicts for #/definitions/{models}
'use_models': False,
# This library has some issues with nullable fields
'validate_responses': False,
# Returns response in 2-tuple of (body, response); if False, will only return body
'also_return_response': True,
}
bitMEX = SwaggerClient.from_url(
SPEC_URI,
config=config)
pp = pprint.PrettyPrinter(indent=2)
# You can get a feel for what is available by printing these objects.
# See also https://testnet.bitmex.com/api/explorer
print('---The BitMEX Object:---')
print(dir(bitMEX))
print('\n---The BitMEX.Trade Object:---')
print(dir(bitMEX.Trade))
# Basic unauthenticated call
res, http_response = bitMEX.Trade.Trade_get(symbol='XBTUSD', count=1).result()
print('\n---A basic Trade GET:---')
pp.pprint(res)
def setup_client(
url: str,
schema: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
"""
:returns: error message, if appropriate.
"""
if get_abstraction().client:
return None
try:
config = {'internally_dereference_refs': True}
if not schema:
client = SwaggerClient.from_url(url, config=config)
else:
client = SwaggerClient.from_spec(schema, origin_url=url, config=config)
except requests.exceptions.ConnectionError:
return 'Unable to connect to server.'
except (
simplejson.errors.JSONDecodeError, # type: ignore
yaml.YAMLError,
HTTPError,
):
return (
'Invalid swagger file. Please check to make sure the '
'swagger file can be found at: {}.'.format(url)
)
except SwaggerValidationError:
return 'Invalid swagger format.'
SPEC_URI = HOST + "/api/explorer/swagger.json"
config = {
'use_models': False,
'validate_responses': False,
'also_return_response': True,
}
bitMEX = SwaggerClient.from_url(
SPEC_URI,
config=config)
self.API_KEY = API_KEY
self.API_SECRET = API_SECRET
request_client = RequestsClient()
print("websocket start 2")
request_client.authenticator = APIKeyAuthenticator(HOST, self.API_KEY, self.API_SECRET)
self.bitMEXAuthenticated = SwaggerClient.from_url(
SPEC_URI,
config=config,
http_client=request_client)
print("websocket end")
# Basic authenticated call
print('\n---A basic Position GET:---')
print('The following call requires an API key. If one is not set, it will throw an Unauthorized error.')
self.avgPrice = 0
self.pos = 0
def order(self,count=1,side="Buy"):