Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def load_trs_client(service_id, http_client=None):
"""
Return an API client for the selected workflow execution service.
"""
if http_client is None:
http_client = _init_http_client(service_id=service_id)
spec_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
'ga4gh-tool-discovery.yaml')
spec_path = os.path.abspath(spec_path)
opts = _get_trs_opts(service_id)
api_url = '{}://{}'.format(opts['proto'], opts['host'])
loader = Loader(http_client, request_headers=None)
spec_dict = loader.load_spec('file:///{}'.format(spec_path),
base_url=api_url)
spec_client = SwaggerClient.from_spec(spec_dict,
origin_url=api_url,
http_client=http_client,
config={'use_models': False})
return spec_client.GA4GH
:param spec_url: url pointing at the swagger API specification
:type spec_url: str
:param http_client: an HTTP client used to perform requests
:type http_client: :class:`bravado.http_client.HttpClient`
:param request_headers: Headers to pass with http requests
:type request_headers: dict
:param config: Config dict for bravado and bravado_core.
See CONFIG_DEFAULTS in :module:`bravado_core.spec`.
See CONFIG_DEFAULTS in :module:`bravado.client`.
:rtype: :class:`bravado_core.spec.Spec`
"""
log.debug(u"Loading from %s", spec_url)
http_client = http_client or RequestsClient()
loader = Loader(http_client, request_headers=request_headers)
spec_dict = loader.load_spec(spec_url)
# RefResolver may have to download additional json files (remote refs)
# via http. Wrap http_client's request() so that request headers are
# passed along with the request transparently. Yeah, this is not ideal,
# but since RefResolver has new found responsibilities, it is
# functional.
if request_headers is not None:
http_client.request = inject_headers_for_remote_refs(
http_client.request, request_headers)
return cls.from_spec(spec_dict, spec_url, http_client, config)
def load(self):
http_client = FidoClientGlobalHeaders(headers=self.headers)
loader = Loader(http_client)
protocol = "https" if self.apikey else "http"
origin_url = f"{protocol}://{self.host}/apispec.json"
spec_dict = loader.load_spec(origin_url)
spec_dict["host"] = self.host
spec_dict["schemes"] = [protocol]
config = {
"validate_responses": False,
"use_models": False,
"include_missing_properties": False,
"formats": [email_format, url_format],
}
bravado_config = bravado_config_from_config_dict(config)
for key in set(bravado_config._fields).intersection(set(config)):
del config[key]
config["bravado"] = bravado_config
def get_swagger_json(spec_uri, exclude_formats=[]):
loader = Loader(RequestsClient())
spec_dict = loader.load_spec(spec_uri)
if not exclude_formats:
return spec_dict
# exlude formats from definitions
for def_key, def_item in spec_dict['definitions'].items():
if 'properties' not in def_item:
continue
for prop_key, prop_item in def_item['properties'].items():
if 'format' in prop_item and prop_item['format'] in exclude_formats:
prop_item.pop('format')
# exlude formats from paths
for path_key, path_item in spec_dict['paths'].items():
for method_key, method_item in path_item.items():
if 'parameters' not in method_item:
def from_url(cls, spec_url, http_client=None):
http_client = http_client or RequestsClient()
loader = Loader(http_client)
swagger_docs = loader.load_spec(spec_url)
return cls(swagger_docs)
'validate_requests': False,
'validate_responses': False
}
url_info = urlparse(url)
server = f'{url_info.scheme}://{url_info.netloc}'
if auth_type == 'basic':
LOG.debug(f'Using basic auth on {server}')
http_client = BasicRealmClient()
http_client.set_realm_basic_auth(
host=url_info.netloc,
username=user,
password=pw,
realm=realm,
)
loader = Loader(http_client, request_headers=None)
try:
spec_url = _SPEC_URL.format(url)
LOG.debug(f'Loading schema from: {spec_url}')
spec_dict = loader.load_spec(spec_url)
except bravado.exception.HTTPForbidden as forb:
LOG.error('Could not authenticate with provided credentials')
raise forb
except (
bravado.exception.HTTPBadGateway,
bravado.exception.BravadoConnectionError
) as bgwe:
LOG.error('Server Unavailable')
raise bgwe
else:
LOG.debug(f'getting OIDC session on realm {realm}')
def config(cls, url, http_client=None, request_headers=None):
"""
Accepts an optionally configured requests client with authentication
details set.
:param url: The URL of the service to connect to
:param http_client: The http_client to use, \
defaults to :func:`RequestsClient`
:param request_headers: The headers to set on each request.
:return:
"""
swagger_path = '{}/swagger.json'.format(url.rstrip('/'))
http_client = http_client or RequestsClient()
loader = Loader(http_client, request_headers=request_headers)
spec_dict = loader.load_spec(swagger_path)
return spec_dict
def load_url(spec_url, http_client=None, base_url=None):
"""Loads a Swagger spec.
:param spec_url: URL for swagger.json.
:param http_client: HTTP client interface.
:param base_url: Optional URL to be the base URL for finding API
declarations. If not specified, 'basePath' from the
resource listing is used.
:return: validated spec in dict form
:raise: IOError, URLError: On error reading api-docs.
"""
if http_client is None:
http_client = RequestsClient()
loader = Loader(http_client=http_client)
return loader.load_spec(spec_url, base_url=base_url)