Add RequestClient class to handle http requests

Refactor request_util.py - Add RequestClient to handle network requests

tweeterpy/api_util.py

@@ -5,7 +5,7 @@
 import demjson3
 import logging.config
 from tweeterpy import config
-from tweeterpy.request_util import make_request
+from tweeterpy.request_util import RequestClient
 from tweeterpy.constants import Path, FeatureSwitch, API_TMP_FILE
 
 logging.config.dictConfig(config.LOGGING_CONFIG)
@@ -24,8 +24,8 @@ class ApiUpdater:
         Twitter updates its API quite frequently. Therefore, ApiUpdater checks for the latest updates and modifies the api_endpoints, feature_switches, path etc in constants.py
     """
 
-    def __init__(self, update_api=True, session=None):
-        self.__session = session
+    def __init__(self, request_client: RequestClient = None, update_api: bool = True):
+        self.request_client = request_client
         try:
             logger.debug('Updating API...')
             # fmt: off - Turns off formatting for this block of code.
@@ -60,7 +60,7 @@ def __init__(self, update_api=True, session=None):
             # fmt: on 
 
     def _get_home_page_source(self):
-        return str(make_request(Path.BASE_URL, session=self.__session))
+        return str(self.request_client.request(Path.BASE_URL))
 
     def _get_api_file_url(self, page_source=None):
         if page_source is None:
@@ -89,12 +89,12 @@ def _get_main_file_url(self, page_source=None):
     def _get_api_file_content(self, file_url=None):
         if file_url is None:
             file_url = self._get_api_file_url()
-        return str(make_request(file_url, session=self.__session))
+        return str(self.request_client.request(file_url))
 
     def _get_main_file_content(self, file_url=None):
         if file_url is None:
             file_url = self._get_main_file_url()
-        return str(make_request(file_url, session=self.__session))
+        return str(self.request_client.request(file_url))
 
     def _js_to_py_dict(sel, page_source):
         if isinstance(page_source, list):

tweeterpy/login_util.py

@@ -1,12 +1,12 @@
 from tweeterpy.constants import Path
 from tweeterpy.util import find_nested_key
-from tweeterpy.request_util import make_request
+from tweeterpy.request_util import RequestClient
 from tweeterpy.logging_util import disable_logger
 
 
 class TaskHandler:
-    def __init__(self, session=None):
-        self.__session = session
+    def __init__(self, request_client: RequestClient = None):
+        self.request_client = request_client
 
     def _create_task_mapper(self, username, password, verification_input_data):
         # fmt: off  - Turns off formatting for this block of code. Just for the readability purpose.
@@ -32,19 +32,19 @@ def _get_flow_token(self):
                                  'phone_verification': 4, 'privacy_options': 1, 'security_key': 3, 'select_avatar': 4, 'select_banner': 2,
                                  'settings_list': 7, 'show_code': 1, 'sign_up': 2, 'sign_up_review': 4, 'tweet_selection_urt': 1, 'update_users': 1,
                                  'upload_media': 1, 'user_recommendations_list': 4, 'user_recommendations_urt': 1, 'wait_spinner': 3, 'web_modal': 1}}
-        return make_request(Path.TASK_URL, method="POST", params=params, json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", params=params, json=payload)
 
     def _get_javscript_instrumentation_subtask(self):
         params = {'c_name': 'ui_metrics'}
-        return make_request(Path.JAVSCRIPT_INSTRUMENTATION_URL, params=params, session=self.__session)
+        return self.request_client.request(Path.JAVSCRIPT_INSTRUMENTATION_URL, params=params)
 
     def _get_user_flow_token(self, flow_token, subtask_id="LoginJsInstrumentationSubtask"):
         payload = {'flow_token': flow_token,
                    'subtask_inputs': [{'subtask_id': subtask_id,
                                       'js_instrumentation': {
                                           'response': '',
                                           'link': 'next_link'}}]}
-        return make_request(Path.TASK_URL, method="POST", json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", json=payload)
 
     @disable_logger
     def _get_password_flow_token(self, flow_token, subtask_id="LoginEnterUserIdentifierSSO", username=None):
@@ -53,31 +53,31 @@ def _get_password_flow_token(self, flow_token, subtask_id="LoginEnterUserIdentif
                                       'settings_list': {
                                           'setting_responses': [{'key': 'user_identifier', 'response_data': {'text_data': {'result': username}}}],
                                           'link': 'next_link'}}]}
-        return make_request(Path.TASK_URL, method="POST", json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", json=payload)
 
     @disable_logger
     def _get_account_duplication_flow_token(self, flow_token, subtask_id="LoginEnterPassword", password=None):
         payload = {'flow_token': flow_token,
                    'subtask_inputs': [{'subtask_id': subtask_id,
                                       'enter_password': {'password': password, 'link': 'next_link'}}]}
-        return make_request(Path.TASK_URL, method="POST", json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", json=payload)
 
     def _check_suspicious_login(self, flow_token, subtask_id="DenyLoginSubtask"):
         payload = {"flow_token": flow_token,
                    "subtask_inputs": [{"subtask_id": subtask_id, "cta": {"link": "next_link"}}]}
-        return make_request(Path.TASK_URL, method="POST", json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", json=payload)
 
     def _check_account_duplication(self, flow_token, subtask_id="AccountDuplicationCheck"):
         payload = {'flow_token': flow_token,
                    'subtask_inputs': [{'subtask_id': subtask_id, 'check_logged_in_account': {'link': 'AccountDuplicationCheck_false'}}]}
-        return make_request(Path.TASK_URL, method="POST", json=payload, session=self.__session)
+        return self.request_client.request(Path.TASK_URL, method="POST", json=payload)
 
     def _handle_suspicious_login(self, flow_token, subtask_id="LoginAcid",verification_input_data=None):
         payload = {"flow_token": flow_token,
                    "subtask_inputs": [{"subtask_id": subtask_id, "enter_text": {"text": verification_input_data,"link":"next_link"}}]}
         handle_incorrect_input = True
         while handle_incorrect_input:
-            response = make_request(Path.TASK_URL, method="POST", json=payload, skip_error_checking=True, session=self.__session)
+            response = self.request_client.request(Path.TASK_URL, method="POST", json=payload, skip_error_checking=True)
             if isinstance(response, dict) and "errors" in response.keys():
                 error_message = "\n".join([error['message'] for error in response['errors']])
                 payload['subtask_inputs'][0]['enter_text']['text'] = str(input(f"{error_message} - Type again ==> "))

tweeterpy/request_util.py

@@ -3,50 +3,60 @@
 import logging.config
 from tweeterpy import util
 from tweeterpy import config
+from requests.exceptions import ProxyError, InvalidProxyURL
 
 logging.config.dictConfig(config.LOGGING_CONFIG)
 logger = logging.getLogger(__name__)
 
 
-def make_request(url, session=None, method=None, max_retries=None, timeout=None, skip_error_checking=False, **kwargs):
-    if session is None:
-        raise NameError("name 'session' is not defined.")
-    if not isinstance(session, requests.Session):
-        raise TypeError(
-            f"Invalid session type. {session} is not a requests.Session Object...")
-    if method is None:
-        method = "GET"
-    if max_retries is None:
-        max_retries = config.MAX_RETRIES or 3
-    if timeout is None:
-        timeout = config.TIMEOUT or 30
-    logger.debug(f"{locals()}")
-    for retry_count, _ in enumerate(range(max_retries), start=1):
-        response_text, api_limit_stats = "", {}
-        try:
-            response = session.request(method, url, timeout=timeout, **kwargs)
-            api_limit_stats = util.check_api_rate_limits(response) or {}
-            if api_limit_stats:
-                config._RATE_LIMIT_STATS = api_limit_stats
-            soup = bs4.BeautifulSoup(response.content, "lxml")
-            if "json" in response.headers["Content-Type"]:
-                if skip_error_checking:
-                    return response.json()
-                return util.check_for_errors(response.json())
-            response_text = "\n".join(
-                [line.strip() for line in soup.text.split("\n") if line.strip()])
-            response.raise_for_status()
-            return soup
-        except KeyboardInterrupt:
-            logger.warn("Keyboard Interruption...")
-            return
-        except Exception as error:
-            logger.debug(f"Retry No. ==> {retry_count}")
-            if retry_count >= max_retries:
-                logger.exception(f"{error}\n{response_text}\n")
-                if api_limit_stats.get('rate_limit_exhausted'):
-                    logger.error(f"Rate Limit Exceeded => {api_limit_stats}")
-                raise error
+class RequestClient:
+    def __init__(self, session: requests.Session):
+        self.session = session
+        self.client_transaction = None
+
+    def request(self, url, method=None, max_retries=None, timeout=None, skip_error_checking=False, **kwargs):
+        if method is None:
+            method = "GET"
+        if max_retries is None:
+            max_retries = config.MAX_RETRIES or 3
+        if timeout is None:
+            timeout = config.TIMEOUT or 30
+        logger.debug(f"{locals()}")
+        for retry_count, _ in enumerate(range(max_retries), start=1):
+            response_text, api_limit_stats = "", {}
+            try:
+                response = self.session.request(
+                    method, url, timeout=timeout, **kwargs)
+                api_limit_stats = util.check_api_rate_limits(response) or {}
+                if "json" in response.headers.get("Content-Type", ""):
+                    response = response.json()
+                    if api_limit_stats:
+                        config._RATE_LIMIT_STATS = api_limit_stats
+                        response.update({"api_rate_limit": api_limit_stats})
+                    if skip_error_checking:
+                        return response
+                    return util.check_for_errors(response)
+                soup = bs4.BeautifulSoup(response.content, "lxml")
+                response_text = "\n".join(
+                    [line.strip() for line in soup.text.split("\n") if line.strip()])
+                response.raise_for_status()
+                return soup
+            except KeyboardInterrupt:
+                logger.warn("Keyboard Interruption...")
+                return
+            except (ProxyError, InvalidProxyURL) as proxy_error:
+                logger.error(f"{proxy_error}")
+                if retry_count >= max_retries:
+                    raise proxy_error
+            except Exception as error:
+                logger.debug(f"Retry No. ==> {retry_count}")
+                if retry_count >= max_retries:
+                    logger.exception(f"{error}\n{response_text}\n")
+                    if api_limit_stats.get('exhausted'):
+                        logger.error(
+                            f"Rate Limit Exceeded => {api_limit_stats}")
+                        raise util.RateLimitError('API Rate Limit Exceeded.')
+                    raise error
 
 
 if __name__ == '__main__':