Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __getattr__(self, name):
"""
:rtype: :class:`CallableOperation`
"""
return CallableOperation(getattr(self.resource, name), self.also_return_response)
path : str
the API route string value, for example:
"/api/resources/vlan-pools/{id}"
Returns
-------
Request
The request instance you can then use to exeute the command.
"""
op = self.client.swagger_spec.get_op_for_request(method, path)
if not op:
raise RuntimeError(
'no command found for (%s, %s)' % (method, path))
return Request(self.client, CallableOperation(op))
def __getattr__(self, name):
fn = CallableOperation(
getattr(self.resource, self._get_full_name(name)),
self.also_return_response)
# It was annoying to constantly call .response().result to get to the most
# valuable data. Also errors were being swallowed by the inner workings of
# Bravado. Wrapping the returned function handles this.
def resultant_function(*args, **kwargs):
# try:
future = fn(*args, **kwargs)
# On debug, show outgoing requests
LOG.debug(show_request(
getattr(self.resource, self._get_full_name(name)),
*args,
**kwargs
))
# This is an attempt to fix an error that only occurs in travis where kernel
def _swagger_operation(self) -> CallableOperation:
return cast(
CallableOperation,
getattr(
getattr(get_abstraction().client, self.tag),
self.operation_id,
),
from bravado.http_future import HttpFuture
from bravado.requests_client import RequestsClient, RequestsFutureAdapter, \
RequestsResponseAdapter
from bravado_core.spec import Spec
from bravado.exception import HTTPInternalServerError, HTTPNotFound, \
HTTPBadRequest, HTTPForbidden
from .exceptions import ESIError, ESINotFound, ESIForbidden, \
ESIAuthorizationError
from .constants import ESI_ENDPOINT, ESI_DATASOURCE, ESI_SWAGGER_CACHE_TIMER, \
ESY_USER_AGENT
log = logging.getLogger(__name__)
class ESICallableOperation(CallableOperation):
"""
Wraps bravado's CallableOpeartion to handle pagination
"""
def __init__(self, operation):
self.operation = operation
self.require_authorization = any(map(lambda spec: 'evesso' in spec,
self.operation.security_specs))
super(ESICallableOperation, self).__init__(operation)
self.paginated = 'page' in operation.params
def __call__(self, _token=None, **op_kwargs):
"""Invoke the actual HTTP request and return a future.
:rtype: :class:`bravado.http_future.HTTPFuture`
"""
def __getattr__(self, name):
"""
:rtype: :class:`CallableOperation`
"""
return CallableOperation(getattr(self.resource, name), self.also_return_response)
import inspect
from collections import defaultdict
from functools import lru_cache
from functools import wraps
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from bravado.client import CallableOperation
PostFuzzHook = Callable[[CallableOperation, Dict[str, Any]], None]
# These are module variables which contain the post-fuzz hooks
# which have been registered. Each global allows fuzz_lightyear
# to get a list applicable to a certain operation or tag.
_POST_FUZZ_HOOKS_BY_OPERATION = defaultdict(set) # type: Dict[str, Set[PostFuzzHook]]
_POST_FUZZ_HOOKS_BY_TAG = defaultdict(set) # type: Dict[str, Set[PostFuzzHook]]
def register_post_fuzz_hook(
hook: PostFuzzHook,
operation_ids: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
) -> None:
"""Adds a post-fuzz hook to fuzz_lightyear's store of post-fuzz
hooks.