Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def index(request):
ctx = RequestContext(request)
ctx['email'] = request.META.get('HTTP_X_CONSUMER_USERNAME')
api_key = request.META.get('HTTP_X_CONSUMER_CUSTOM_ID')
if api_key and ctx['email']:
http_client.set_api_key(
os.environ['MPCONTRIBS_API_HOST'], b64decode(api_key),
param_in='header', param_name='x-api-key'
)
client = SwaggerClient.from_spec(
spec_dict, http_client=http_client,
config={'validate_responses': False}
)
ctx['landing_pages'] = []
provenances = client.provenances.get_provenances().response().result
for provenance in provenances:
explorer = 'mpcontribs_users_{}_explorer_index'.format(provenance.project)
entry = {'project': provenance.project, 'url': '#'}# TODO reverse(explorer)}
entry['title'] = provenance.title
authors = provenance.authors.split(',', 1)
prov_display = '<span style="font-size: 13px;" class="pull-right">{}'.format(authors[0])
if len(authors) > 1:
prov_display += '''<button style="padding: 0px 0px 2px 5px;" title="{}" data-container="body" data-placement="bottom" data-toggle="tooltip" data-html="true" class="btn btn-sm btn-link">et al.'''.format(
authors[1].strip().replace(', ', '<br>'))</button></span>
def get_client(
url: str,
schema: Optional[Dict[str, Any]] = None,
):
"""
:param url: server URL
"""
if not schema:
return SwaggerClient.from_url(url)
return SwaggerClient.from_spec(
schema,
origin_url=url,
)
log.error("paasta-api swagger spec %s does not exist", swagger_file)
return None
with open(swagger_file) as f:
spec_dict = json.load(f)
# replace localhost in swagger.json with actual api server
spec_dict["host"] = api_server
spec_dict["schemes"] = [parsed.scheme]
# sometimes we want the status code
requests_client = PaastaRequestsClient(
scheme=parsed.scheme, cluster=cluster, system_paasta_config=system_paasta_config
)
if http_res:
config = {"also_return_response": True}
c = SwaggerClient.from_spec(
spec_dict=spec_dict, config=config, http_client=requests_client
)
else:
c = SwaggerClient.from_spec(spec_dict=spec_dict, http_client=requests_client)
return paasta_tools.api.auth_decorator.AuthClientDecorator(c, cluster_name=cluster)
def connect(self):
response = requests.get(
'https://%s:%s/swagger.json' % (self.host, self.port),
verify=self.verify)
if response.status_code == requests.status_codes.codes.OK:
spec = response.json()
spec['host'] = '%s:%s' % (self.host, self.port)
http_client = RequestsClient()
http_client.session.verify = self.verify
config = {
'validate_swagger_spec': True
}
self.client = SwaggerClient.from_spec(spec,
http_client=http_client,
config=config)
else:
raise Exception(response)
# '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)
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.'
get_abstraction().client = client
return None
spec_dict = json.load(f)
# replace localhost in swagger.json with actual api server
spec_dict["host"] = api_server
spec_dict["schemes"] = [parsed.scheme]
# sometimes we want the status code
requests_client = PaastaRequestsClient(
scheme=parsed.scheme, cluster=cluster, system_paasta_config=system_paasta_config
)
if http_res:
config = {"also_return_response": True}
c = SwaggerClient.from_spec(
spec_dict=spec_dict, config=config, http_client=requests_client
)
else:
c = SwaggerClient.from_spec(spec_dict=spec_dict, http_client=requests_client)
return paasta_tools.api.auth_decorator.AuthClientDecorator(c, cluster_name=cluster)