Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_handle_response(self, mocker):
converter = mocker.Mock()
input_value = "converted"
converter.return_value = input_value
rc = hooks.ResponseHandler(converter)
assert rc.handle_response(None) is input_value
def test_delegate_handle_response_multiple(self, mocker):
# Include one hook that can't handle responses
mock_response_handler = mocker.stub()
mock_request_auditor = mocker.stub()
chain = hooks.TransactionHookChain(
hooks.RequestAuditor(mock_request_auditor),
hooks.ResponseHandler(mock_response_handler),
hooks.ResponseHandler(mock_response_handler),
)
chain.handle_response("consumer", {})
mock_response_handler.call_count == 2
mock_request_auditor.call_count == 1
def test_delegate_handle_response_multiple(self, mocker):
# Include one hook that can't handle responses
mock_response_handler = mocker.stub()
mock_request_auditor = mocker.stub()
chain = hooks.TransactionHookChain(
hooks.RequestAuditor(mock_request_auditor),
hooks.ResponseHandler(mock_response_handler),
hooks.ResponseHandler(mock_response_handler),
)
chain.handle_response("consumer", {})
mock_response_handler.call_count == 2
mock_request_auditor.call_count == 1
def test_equals(self):
assert isinstance(
self.type_cls().with_value("hello"), hooks.TransactionHook
)
class _InjectableMethodAnnotation(MethodAnnotation):
def modify_request(self, request_builder):
request_builder.add_transaction_hook(self)
class _BaseHandlerAnnotation(_InjectableMethodAnnotation):
def __new__(cls, func=None, *args, **kwargs):
if func is None:
return lambda f: cls(f, *args, **kwargs)
self = super(_BaseHandlerAnnotation, cls).__new__(cls)
functools.update_wrapper(self, func)
return self
# noinspection PyPep8Naming
class response_handler(_BaseHandlerAnnotation, hooks.ResponseHandler):
"""
A decorator for creating custom response handlers.
To register a function as a custom response handler, decorate the
function with this class. The decorated function should accept a single
positional argument, an HTTP response object:
Example:
.. code-block:: python
@response_handler
def raise_for_status(response):
response.raise_for_status()
return response
Then, to apply custom response handling to a request method, simply
the registered callback should accept a reference to the
:class:`~Consumer` instance as its leading argument:
Example:
.. code-block:: python
@response_handler(requires_consumer=True)
def raise_for_status(consumer, response):
...
.. versionadded:: 0.4.0
"""
# noinspection PyPep8Naming
class error_handler(_BaseHandlerAnnotation, hooks.ExceptionHandler):
"""
A decorator for creating custom error handlers.
To register a function as a custom error handler, decorate the
function with this class. The decorated function should accept three
positional arguments: (1) the type of the exception, (2) the
exception instance raised, and (3) a traceback instance.
Example:
.. code-block:: python
@error_handler
def raise_api_error(exc_type, exc_val, exc_tb):
# wrap client error with custom API error
...
def __init__(self, builder, consumer=None):
self._client = builder.client
self._base_url = str(builder.base_url)
self._converters = list(builder.converters)
self._auth = builder.auth
self._consumer = consumer
if builder.hooks:
self._session_chain = hooks_.TransactionHookChain(*builder.hooks)
else:
self._session_chain = None
def _get_request_hooks(contract):
chain = list(contract.transaction_hooks)
if callable(contract.return_type):
chain.append(hooks_.ResponseHandler(contract.return_type))
return chain
@error_handler(requires_consumer=True)
def raise_api_error(consumer, exc_type, exc_val, exc_tb):
...
.. versionadded:: 0.4.0
Note:
Error handlers can not completely suppress exceptions. The
original exception is thrown if the error handler doesn't throw
anything.
"""
# noinspection PyPep8Naming
class inject(_InjectableMethodAnnotation, hooks.TransactionHookChain):
"""
A decorator that applies one or more hooks to a request method.
def __init__(
self,
base_url="",
client=None,
converters=(),
auth=None,
hooks=(),
**kwargs
):
builder = Builder()
builder.base_url = base_url
builder.converters = kwargs.pop("converter", converters)
hooks = kwargs.pop("hook", hooks)
if isinstance(hooks, hooks_.TransactionHook):
hooks = (hooks,)
builder.add_hook(*hooks)
builder.auth = auth
builder.client = client
self.__session = session.Session(builder)
self.__client = builder.client