refactor: use CloudantBaseService as a super of CloudantV1

ibmcloudant/__init__.py

@@ -1,5 +1,5 @@
 # coding: utf-8
-# © Copyright IBM Corporation 2020, 2024.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -19,23 +19,10 @@
 from ibm_cloud_sdk_core import IAMTokenManager, DetailedResponse, BaseService, ApiException, get_authenticator
 from .couchdb_session_authenticator import CouchDbSessionAuthenticator
 from .couchdb_session_get_authenticator_patch import new_construct_authenticator
-from .cloudant_base_service import new_init, new_prepare_request, new_set_default_headers, new_set_http_client, new_set_service_url, new_set_disable_ssl_verification
 from .couchdb_session_token_manager import CouchDbSessionTokenManager
 from .cloudant_v1 import CloudantV1
 from .features.changes_follower import ChangesFollower
 from .features.pagination import Pager, PagerType, Pagination
 
 # sdk-core's __construct_authenticator works with a long switch-case so monkey-patching is required
 get_authenticator.__construct_authenticator = new_construct_authenticator
-
-CloudantV1.__init__ = new_init
-
-CloudantV1.set_service_url = new_set_service_url
-
-CloudantV1.set_default_headers = new_set_default_headers
-
-CloudantV1.prepare_request = new_prepare_request
-
-CloudantV1.set_http_client = new_set_http_client
-
-CloudantV1.set_disable_ssl_verification = new_set_disable_ssl_verification

ibmcloudant/cloudant_base_service.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2024.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -24,12 +24,11 @@
 from json.decoder import JSONDecodeError
 from io import BytesIO
 
+from ibm_cloud_sdk_core import BaseService
 from ibm_cloud_sdk_core.authenticators import Authenticator
 from requests import Response, Session
-from requests.cookies import RequestsCookieJar
 
 from .common import get_sdk_headers
-from .cloudant_v1 import CloudantV1
 from .couchdb_session_authenticator import CouchDbSessionAuthenticator
 
 # pylint: disable=missing-docstring
@@ -72,52 +71,100 @@ def __hash__(self):
         # Since Py3.6 dict is ordered so use a key only dict for our set
         rules_by_operation.setdefault(operation_id, dict()).setdefault(rule)
 
-_old_init = CloudantV1.__init__
-
-def new_init(self, authenticator: Authenticator = None):
-    _old_init(self, authenticator)
-    # Overwrite default read timeout to 2.5 minutes
-    if not ('timeout' in self.http_config):
-        new_http_config = self.http_config.copy()
-        new_http_config['timeout'] = (CONNECT_TIMEOUT, READ_TIMEOUT)
-        self.set_http_config(new_http_config)
-    # Custom actions for CouchDbSessionAuthenticator
-    if isinstance(authenticator, CouchDbSessionAuthenticator):
-        # 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
-    add_hooks(self)
-
-_old_set_service_url = CloudantV1.set_service_url
-
-def new_set_service_url(self, service_url: str):
-    _old_set_service_url(self, service_url)
-    try:
+class CloudantBaseService(BaseService):
+    """
+    The base class for service classes.
+    """
+    def __init__(
+        self,
+        service_url: str = None,
+        authenticator: Authenticator = None,
+    ) -> None:
+        """
+        Construct a new client for the Cloudant service.
+
+        :param Authenticator authenticator: The authenticator specifies the authentication mechanism.
+               Get up to date information from https://github.com/IBM/python-sdk-core/blob/main/README.md
+               about initializing the authenticator of your choice.
+        """
+        BaseService.__init__(self, service_url=service_url, authenticator=authenticator)
+        # Overwrite default read timeout to 2.5 minutes
+        if not ('timeout' in self.http_config):
+            new_http_config = self.http_config.copy()
+            new_http_config['timeout'] = (CONNECT_TIMEOUT, READ_TIMEOUT)
+            self.set_http_config(new_http_config)
+        # Custom actions for CouchDbSessionAuthenticator
+        if isinstance(authenticator, CouchDbSessionAuthenticator):
+            # Make token manager of CouchDbSessionAuthenticator to use the same http client as main service
+            self.authenticator._set_http_client(self.get_http_client(), self.jar)
+        add_hooks(self)
+
+    def set_service_url(self, service_url: str):
+        super().set_service_url(service_url)
+        try:
+            if isinstance(self.authenticator, CouchDbSessionAuthenticator):
+                self.authenticator.token_manager.set_service_url(service_url)
+        except AttributeError:
+            pass  # in case no authenticator is configured yet, pass
+
+    def set_default_headers(self, headers: Dict[str, str]):
+        super().set_default_headers(headers)
+        if isinstance(self.authenticator, CouchDbSessionAuthenticator):
+            combined_headers = {}
+            combined_headers.update(headers)
+            combined_headers.update(get_sdk_headers(
+                service_name=self.DEFAULT_SERVICE_NAME,
+                service_version='V1',
+                operation_id='authenticator_post_session')
+            )
+            self.authenticator.token_manager.set_default_headers(combined_headers)
+
+    def set_disable_ssl_verification(self, status: bool = False) -> None:
+        super().set_disable_ssl_verification(status)
+        if isinstance(self.authenticator, CouchDbSessionAuthenticator):
+            self.authenticator.token_manager.set_disable_ssl_verification(status)
+
+    def set_http_client(self, http_client: Session) -> None:
+        super().set_http_client(http_client)
         if isinstance(self.authenticator, CouchDbSessionAuthenticator):
-            self.authenticator.token_manager.set_service_url(service_url)
-    except AttributeError:
-        pass  # in case no authenticator is configured yet, pass
-
-_old_set_default_headers = CloudantV1.set_default_headers
-
-def new_set_default_headers(self, headers: Dict[str, str]):
-    _old_set_default_headers(self, headers)
-    if isinstance(self.authenticator, CouchDbSessionAuthenticator):
-        combined_headers = {}
-        combined_headers.update(headers)
-        combined_headers.update(get_sdk_headers(
-            service_name=self.DEFAULT_SERVICE_NAME,
-            service_version='V1',
-            operation_id='authenticator_post_session')
-        )
-        self.authenticator.token_manager.set_default_headers(combined_headers)
-
-_old_set_disable_ssl_verification = CloudantV1.set_disable_ssl_verification
-
-def new_set_disable_ssl_verification(self, status: bool = False) -> None:
-    _old_set_disable_ssl_verification(self, status)
-    if isinstance(self.authenticator, CouchDbSessionAuthenticator):
-        self.authenticator.token_manager.set_disable_ssl_verification(status)
+            self.authenticator._set_http_client(self.get_http_client(), self.jar)
+        add_hooks(self)
+
+    def prepare_request(self,
+                            method: str,
+                            url: str,
+                            *args,
+                            headers: Optional[dict] = None,
+                            params: Optional[dict] = None,
+                            data: Optional[Union[str, dict]] = None,
+                            files: Optional[Union[Dict[str, Tuple[str]],
+                                                List[Tuple[str,
+                                                            Tuple[str,
+                                                                ...]]]]] = None,
+                            **kwargs) -> dict:
+        # Extract the operation ID from the request headers.
+        operation_id = None
+        header = headers.get('X-IBMCloud-SDK-Analytics')
+        if header is not None:
+            for element in header.split(';'):
+                if element.startswith('operation_id'):
+                    operation_id = element.split('=')[1]
+                    break
+        if operation_id is not None:
+            # Check each validation rule that applies to the operation.
+            # Until the request URL is passed to old_prepare_request it does not include the
+            # service URL and is relative to it
+            request_url_path_segments = urlsplit(url).path.strip('/').split('/')
+            if len(request_url_path_segments) == 1 and request_url_path_segments[0] == '':
+                request_url_path_segments = []
+            # Note the get returns a value-less dict, we are iterating only the keys
+            for rule in rules_by_operation.get(operation_id, {}):
+                if len(request_url_path_segments) > rule.path_segment_index:
+                    segment_to_validate = request_url_path_segments[rule.path_segment_index]
+                    if segment_to_validate.startswith('_'):
+                        raise ValueError('{0} {1} starts with the invalid _ character.'.format(rule.error_parameter_name,
+                            unquote(segment_to_validate)))
+        return super().prepare_request(method, url, *args, headers=headers, params=params, data=data, files=files, **kwargs)
 
 def _error_response_hook(response:Response, *args, **kwargs) -> Optional[Response]:
     # pylint: disable=W0613
@@ -186,52 +233,7 @@ def _error_response_hook(response:Response, *args, **kwargs) -> Optional[Respons
                 # so the exception can surface elsewhere.
                 pass
     return response
-
-_old_prepare_request = CloudantV1.prepare_request
-
-def new_prepare_request(self,
-                        method: str,
-                        url: str,
-                        *args,
-                        headers: Optional[dict] = None,
-                        params: Optional[dict] = None,
-                        data: Optional[Union[str, dict]] = None,
-                        files: Optional[Union[Dict[str, Tuple[str]],
-                                              List[Tuple[str,
-                                                         Tuple[str,
-                                                               ...]]]]] = None,
-                        **kwargs) -> dict:
-    # Extract the operation ID from the request headers.
-    operation_id = None
-    header = headers.get('X-IBMCloud-SDK-Analytics')
-    if header is not None:
-        for element in header.split(';'):
-            if element.startswith('operation_id'):
-                operation_id = element.split('=')[1]
-                break
-    if operation_id is not None:
-        # Check each validation rule that applies to the operation.
-        # Until the request URL is passed to old_prepare_request it does not include the
-        # service URL and is relative to it
-        request_url_path_segments = urlsplit(url).path.strip('/').split('/')
-        if len(request_url_path_segments) == 1 and request_url_path_segments[0] == '':
-            request_url_path_segments = []
-        # Note the get returns a value-less dict, we are iterating only the keys
-        for rule in rules_by_operation.get(operation_id, {}):
-            if len(request_url_path_segments) > rule.path_segment_index:
-                segment_to_validate = request_url_path_segments[rule.path_segment_index]
-                if segment_to_validate.startswith('_'):
-                    raise ValueError('{0} {1} starts with the invalid _ character.'.format(rule.error_parameter_name,
-                        unquote(segment_to_validate)))
-    return _old_prepare_request(self, method, url, *args, headers=headers, params=params, data=data, files=files, **kwargs)
-
 def add_hooks(self):
     response_hooks = self.get_http_client().hooks['response']
     if _error_response_hook not in response_hooks:
         response_hooks.append(_error_response_hook)
-
-_old_set_http_client = CloudantV1.set_http_client
-
-def new_set_http_client(self, http_client: Session) -> None:
-    _old_set_http_client(self, http_client)
-    add_hooks(self)

ibmcloudant/cloudant_v1.py

@@ -27,19 +27,20 @@
 import json
 import logging
 
-from ibm_cloud_sdk_core import BaseService, DetailedResponse
+from ibm_cloud_sdk_core import DetailedResponse
 from ibm_cloud_sdk_core.authenticators.authenticator import Authenticator
 from ibm_cloud_sdk_core.get_authenticator import get_authenticator_from_environment
 from ibm_cloud_sdk_core.utils import convert_list, convert_model, datetime_to_string, string_to_datetime
 
+from .cloudant_base_service import CloudantBaseService
 from .common import get_sdk_headers
 
 ##############################################################################
 # Service
 ##############################################################################
 
 
-class CloudantV1(BaseService):
+class CloudantV1(CloudantBaseService):
     """The Cloudant V1 service."""
 
     DEFAULT_SERVICE_URL = 'https://~replace-with-cloudant-host~.cloudantnosqldb.appdomain.cloud'
@@ -72,7 +73,7 @@ def __init__(
                Get up to date information from https://github.com/IBM/python-sdk-core/blob/main/README.md
                about initializing the authenticator of your choice.
         """
-        BaseService.__init__(self, service_url=self.DEFAULT_SERVICE_URL, authenticator=authenticator)
+        CloudantBaseService.__init__(self, service_url=self.DEFAULT_SERVICE_URL, authenticator=authenticator)
         # enable gzip compression of request bodies
         self.set_enable_gzip_compression(True)
 

ibmcloudant/couchdb_session_authenticator.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2022.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,7 +16,9 @@
 """
 Module for handling session authentication
 """
-from requests import Request
+from requests import Request, Session
+from requests.cookies import RequestsCookieJar
+
 
 from ibm_cloud_sdk_core.authenticators import Authenticator
 from .couchdb_session_token_manager import CouchDbSessionTokenManager
@@ -46,20 +48,22 @@ def __init__(self,
         if not isinstance(disable_ssl_verification, bool):
             raise TypeError('disable_ssl_verification must be a bool')
 
-        self.jar = None
-
         self.token_manager = CouchDbSessionTokenManager(
             username,
             password,
             disable_ssl_verification=disable_ssl_verification
         )
         self.validate()
 
-    def set_jar(self, jar):
-        """Sets the cookie jar for the authenticator.
+    def _set_http_client(self, http_client: Session, jar: RequestsCookieJar) -> None:
+        """Sets base serivice's http client for the authenticator.
         This is an internal method called by BaseService. Not to be called directly.
         """
-        self.jar = jar
+        if isinstance(http_client, Session):
+            self.token_manager.http_client = http_client
+            self.token_manager.jar = jar
+        else:
+            raise TypeError("http_client parameter must be a requests.sessions.Session")
 
     def validate(self):
         """Validates the username, and password for session token requests.
@@ -82,11 +86,7 @@ def authenticate(self, req: Request):
         Args:
             req: Ignored. BaseService uses the cookie jar for every request
         """
-        jar = self.token_manager.get_token()
-        # Requests seem to save cookies only for Sessions. BaseService is
-        # hard-coded to work with "regular" requests requests so updating
-        # the jar manually is necessary
-        self.jar.update(jar)
+        self.token_manager.get_token()
 
     def authentication_type(self) -> str:
         """Returns this authenticator's type ('COUCHDB_SESSION')."""

ibmcloudant/couchdb_session_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2022.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 """
 Module for managing session authentication token
 """
+from requests import Session
+
 from ibm_cloud_sdk_core.token_managers.token_manager import TokenManager
 
 
@@ -52,7 +54,9 @@ def __init__(self, username: str, password: str,
 
         self.token = None
 
+        self.http_client = None
         self.http_config = {}
+        self.jar = None
         self.headers = None
 
     def request_token(self):
@@ -63,14 +67,19 @@ def request_token(self):
              A CookieJar of Cookies the server sent back.
         """
 
-        response = self._request(
+        if not isinstance(self.http_client, Session):
+            raise TypeError("http_client parameter must be a requests.sessions.Session")
+
+        response = self.http_client.request(
             method='POST',
             url=self.url + "/_session",
             headers=self.headers,
             json={
                 'username': self.username,
                 'password': self.password,
-            }
+            },
+            cookies=self.jar,
+            **self.http_config
         )
 
         return response