Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_kafka_discovery_client(client_id):
"""Create smartstack-client for kafka_discovery service."""
# Default retry is 1 on response timeout
retry_config = UserFacingRetryConfig(timeout=RESPONSE_TIMEOUT)
swagger_url = get_swagger_url()
swagger_client = SwaggerClient.from_url(
swagger_url,
RequestsClient(),
)
zipkin_wrapped_client = ZipkinClientDecorator(swagger_client)
return SmartStackClient(
zipkin_wrapped_client,
retry_config,
client_name=client_id,
service_name='kafka_discovery',
)
def _get_swagger_client(self, url):
return SwaggerClient.from_url(
url,
config=dict(
validate_swagger_spec=False,
validate_requests=False,
validate_responses=False,
formats=[uuid_format]
),
http_client=self._http_client
)
def __init__(self, api_address, api_token):
self.api_address = api_address
self.api_token = api_token
http_client = RequestsClient()
self.backend_swagger_client = SwaggerClient.from_url(
'{}/api/backend/swagger.json'.format(self.api_address),
config=dict(
validate_swagger_spec=False,
formats=[uuid_format]),
http_client=http_client)
self.leaderboard_swagger_client = SwaggerClient.from_url(
'{}/api/leaderboard/swagger.json'.format(self.api_address),
config=dict(
validate_swagger_spec=False,
validate_responses=False, # TODO!!!
formats=[uuid_format]),
http_client=http_client)
http_client.authenticator = NeptuneAuthenticator(
self.backend_swagger_client.api.exchangeApiToken(X_Neptune_Api_Token=api_token).response().result)
def __init__(
self, url, config=DEFAULT_CONFIG,
http_client=None, request_headers=None):
swagger_path = '{}/swagger.json'.format(url.rstrip("/"))
config['formats'] = [int64_format]
self._config = config
self.models = SwaggerClient.from_url(swagger_path,
config=config,
http_client=http_client,
request_headers=request_headers)
self.client = self.models.WorkflowExecutionService
def __init__(self, api_address, api_token):
self.api_address = api_address
self.api_token = api_token
http_client = RequestsClient()
self.backend_swagger_client = SwaggerClient.from_url(
'{}/api/backend/swagger.json'.format(self.api_address),
config=dict(
validate_swagger_spec=False,
formats=[uuid_format]),
http_client=http_client)
self.leaderboard_swagger_client = SwaggerClient.from_url(
'{}/api/leaderboard/swagger.json'.format(self.api_address),
config=dict(
validate_swagger_spec=False,
validate_responses=False, # TODO!!!
formats=[uuid_format]),
http_client=http_client)
http_client.authenticator = NeptuneAuthenticator(
self.backend_swagger_client.api.exchangeApiToken(X_Neptune_Api_Token=api_token).response().result)
# 'validate_swagger_spec': True,
# 'validate_requests': True,
# 'formats': [],
}
host = 'https://www.bitmex.com'
if test:
host = 'https://testnet.bitmex.com'
spec_uri = host + '/api/explorer/swagger.json'
spec_dict = get_swagger_json(spec_uri, exclude_formats=EXCLUDE_SWG_FORMATS)
if api_key and api_secret:
request_client = RequestsClient()
request_client.authenticator = APIKeyAuthenticator(host, api_key, api_secret)
return SwaggerClient.from_spec(spec_dict, origin_url=spec_uri, http_client=request_client, config=config)
else:
return SwaggerClient.from_spec(spec_dict, origin_url=spec_uri, http_client=None, config=config)
ResourceDecorator,
CallableOperation,
construct_request
)
from bravado.config import bravado_config_from_config_dict
from bravado.swagger_model import Loader
from .exceptions import AetherAPIException
from .basic_auth import BasicRealmClient
from .oidc import OauthClient
from .logger import LOG
_SPEC_URL = '{}/v1/schema/?format=openapi'
class Client(SwaggerClient):
AUTH_METHODS = ['oauth', 'basic']
def __init__(
self,
url,
user=None,
pw=None,
offline_token=None,
log_level='ERROR',
config=None,
realm=None,
# if using a gateway from a non-standard location
keycloak_url=None,
auth_type='oauth',
# used to specify gateway endpoint ({realm}/{endpoint_name})
endpoint_name='kernel',