feat: improve error responses

Create _error_response_hook function to augment
error messages.
Add hook to the client requests session.
Unit tests for the hook.

Author: ricellis

ibmcloudant/__init__.py

@@ -1,5 +1,5 @@
 # coding: utf-8
-# © Copyright IBM Corporation 2020, 2021.
+# © Copyright IBM Corporation 2020, 2024.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

ibmcloudant/cloudant_base_service.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2022.
+# © Copyright IBM Corporation 2020, 2024.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,12 +15,17 @@
 # limitations under the License.
 """
 Module to patch sdk core base service for session authentication
+and other helpful features.
 """
 from collections import namedtuple
 from typing import Dict, Optional, Union, Tuple, List
 from urllib.parse import urlsplit, unquote
+from json import dumps
+from json.decoder import JSONDecodeError
+from io import BytesIO
 
 from ibm_cloud_sdk_core.authenticators import Authenticator
+from requests import Response
 from requests.cookies import RequestsCookieJar
 
 from .common import get_sdk_headers
@@ -83,7 +88,9 @@ def new_init(self, authenticator: Authenticator = None):
         # Replacing BaseService's http.cookiejar.CookieJar as RequestsCookieJar supports update(CookieJar)
         self.jar = RequestsCookieJar(self.jar)
         self.authenticator.set_jar(self.jar)  # Authenticators don't have access to cookie jars by default
-
+    response_hooks = self.get_http_client().hooks['response']
+    if _error_response_hook not in response_hooks:
+        response_hooks.append(_error_response_hook)
 
 old_set_service_url = CloudantV1.set_service_url
 
@@ -121,10 +128,73 @@ def new_set_disable_ssl_verification(self, status: bool = False) -> None:
     if isinstance(self.authenticator, CouchDbSessionAuthenticator):
         self.authenticator.token_manager.set_disable_ssl_verification(status)
 
+def _error_response_hook(response:Response, *args, **kwargs) -> Optional[Response]:
+    # pylint: disable=W0613
+    # unused args and kwargs required by requests event hook interface
+    """Function for augmenting error responses.
+    Converts the Cloudant response to better match the
+    standard error response formats including adding a
+    trace ID and appending the Cloudant/CouchDB error
+    reason to the message.
+
+    Follows the requests event hook pattern.
+
+    :param response: the requests Response object
+    :type response: Response
+
+    :return: A new response object, defaults to the existing response
+    :rtype: Response,optional
+    """
+    # Only hook into error responses
+    # Ignore HEAD request responses because there is no body to read
+    if not response.ok and response.request.method != 'HEAD':
+        content_type = response.headers.get('content-type')
+        # If it isn't JSON don't mess with it!
+        if content_type is not None and content_type.startswith('application/json'):
+            try:
+                error_json: dict = response.json()
+                # Only augment if there isn't a trace or errors already
+                send_augmented_response = False
+                if 'trace' not in error_json:
+                    if 'errors' not in error_json:
+                        error = error_json.get('error')
+                        reason = error_json.get('reason')
+                        if error is not None:
+                            error_model: dict = {'code': error, 'message': f'{error}'}
+                            if reason:
+                                error_model['message'] += f': {reason}'
+                            error_json['errors'] = [error_model]
+                            send_augmented_response = True
+                    if 'errors' in error_json:
+                        trace = response.headers.get('x-couch-request-id')
+                        if trace is not None:
+                            # Augment trace if there was a value
+                            error_json['trace'] = trace
+                            send_augmented_response = True
+                    if send_augmented_response:
+                        # It'd be nice to just change content on response, but it's internal.
+                        # Instead copy the named attributes to a new Response and then set
+                        # the encoding and bytes of the modified error body.
+                        error_response = Response()
+                        error_response.status_code = response.status_code
+                        error_response.headers = response.headers
+                        error_response.url = response.url
+                        error_response.history = response.history
+                        error_response.reason = response.reason
+                        error_response.cookies = response.cookies
+                        error_response.elapsed = response.elapsed
+                        error_response.request = response.request
+                        error_response.encoding = 'utf-8'
+                        error_response.raw = BytesIO(dumps(error_json).encode('utf-8'))
+                        return error_response
+            except JSONDecodeError:
+                # If we couldn't read the JSON we just return the response as-is
+                # so the exception can surface elsewhere.
+                pass
+    return response
 
 old_prepare_request = CloudantV1.prepare_request
 
-
 def new_prepare_request(self,
                         method: str,
                         url: str,