Retries a 401 error by refreshing the access token

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 2.4.0
+  * Added Retry and exception handling for 401 errors
+
 ## 2.3.1
   * Add new fields into the schema of credit_notes stream [#117](https://github.com/singer-io/tap-xero/pull/117)
 

setup.py

@@ -2,7 +2,7 @@
 from setuptools import setup, find_packages
 
 setup(name="tap-xero",
-      version="2.3.1",
+      version="2.4.0",
       description="Singer.io tap for extracting data from the Xero API",
       author="Stitch",
       url="http://singer.io",

tap_xero/__init__.py

@@ -64,8 +64,8 @@ def load_metadata(stream, schema):
     return metadata.to_list(mdata)
 
 
-def ensure_credentials_are_valid(config):
-    XeroClient(config).filter("currencies")
+def ensure_credentials_are_valid(config, config_path):
+    XeroClient(config, config_path).filter("currencies")
 
 def discover(ctx):
     ctx.check_platform_access()

tap_xero/client.py

@@ -178,13 +178,18 @@ def retry_after_wait_gen():
         yield math.floor(float(sleep_time_str))
 
 class XeroClient():
-    def __init__(self, config):
+    def __init__(self, config, config_path):
         self.session = requests.Session()
         self.user_agent = config.get("user_agent")
+        self.config = config
+        self.config_path = config_path
         self.tenant_id = None
         self.access_token = None
 
-    def refresh_credentials(self, config, config_path):
+    def refresh_credentials(self):
+
+        config = self.config
+        config_path = self.config_path
 
         header_token = b64encode((config["client_id"] + ":" + config["client_secret"]).encode('utf-8'))
 
@@ -205,18 +210,22 @@ def refresh_credentials(self, config, config_path):
             resp = resp.json()
 
             # Write to config file
+            self.access_token = resp["access_token"]
+            self.tenant_id = config['tenant_id']
             config['refresh_token'] = resp["refresh_token"]
+            config['access_token'] = resp["access_token"]
+            self.config = config
             update_config_file(config, config_path)
-            self.access_token = resp["access_token"]
             self.tenant_id = config['tenant_id']
+            LOGGER.info("access_token refreshed")
 
 
     @backoff.on_exception(backoff.expo, (json.decoder.JSONDecodeError, XeroInternalError), max_tries=3)
     @backoff.on_exception(retry_after_wait_gen, XeroTooManyInMinuteError, giveup=is_not_status_code_fn([429]), jitter=None, max_tries=3)
-    def check_platform_access(self, config, config_path):
+    def check_platform_access(self):
 
         # Validating the authentication of the provided configuration
-        self.refresh_credentials(config, config_path)
+        self.refresh_credentials()
 
         headers = {
             "Authorization": "Bearer " + self.access_token,
@@ -233,7 +242,7 @@ def check_platform_access(self, config, config_path):
             raise_for_error(response)
 
 
-    @backoff.on_exception(backoff.expo, (json.decoder.JSONDecodeError, XeroInternalError), max_tries=3)
+    @backoff.on_exception(backoff.expo, (json.decoder.JSONDecodeError, XeroInternalError, XeroUnauthorizedError), max_tries=3)
     @backoff.on_exception(retry_after_wait_gen, XeroTooManyInMinuteError, giveup=is_not_status_code_fn([429]), jitter=None, max_tries=3)
     def filter(self, tap_stream_id, since=None, **params):
         xero_resource_name = tap_stream_id.title().replace("_", "")
@@ -249,6 +258,10 @@ def filter(self, tap_stream_id, since=None, **params):
         request = requests.Request("GET", url, headers=headers, params=params)
         response = self.session.send(request.prepare())
 
+        if response.status_code == 401:
+            self.refresh_credentials()
+            raise_for_error(response)
+
         if response.status_code != 200:
             raise_for_error(response)
             return None

tap_xero/context.py

@@ -9,13 +9,13 @@ def __init__(self, config, state, catalog, config_path):
         self.config_path = config_path
         self.state = state
         self.catalog = catalog
-        self.client = XeroClient(config)
+        self.client = XeroClient(config, config_path=config_path)
 
     def refresh_credentials(self):
-        self.client.refresh_credentials(self.config, self.config_path)
+        self.client.refresh_credentials()
 
     def check_platform_access(self):
-        self.client.check_platform_access(self.config, self.config_path)
+        self.client.check_platform_access()
 
     def get_bookmark(self, path):
         return bks_.get_bookmark(self.state, *path)