diff --git a/CHANGELOG.md b/CHANGELOG.md index 99c5149..c8982ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ # Changelog +## 2.4.0 + * Add backoff and error handling for 401 errors [#121](https://github.com/singer-io/tap-xero/pull/121) + ## 2.3.2 * Bump dependency versions for twistlock compliance [#122](https://github.com/singer-io/tap-xero/pull/122) diff --git a/setup.py b/setup.py index 0b8cc6e..831c503 100755 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup(name="tap-xero", - version="2.3.2", + version="2.4.0", description="Singer.io tap for extracting data from the Xero API", author="Stitch", url="http://singer.io", diff --git a/tap_xero/__init__.py b/tap_xero/__init__.py index 529790a..49355ce 100644 --- a/tap_xero/__init__.py +++ b/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() diff --git a/tap_xero/client.py b/tap_xero/client.py index 17386d5..47253f4 100644 --- a/tap_xero/client.py +++ b/tap_xero/client.py @@ -178,15 +178,17 @@ 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.tenant_id = None + self.config = config + self.config_path = config_path self.access_token = None - def refresh_credentials(self, config, config_path): + def refresh_credentials(self): - header_token = b64encode((config["client_id"] + ":" + config["client_secret"]).encode('utf-8')) + header_token = b64encode((self.config["client_id"] + ":" + self.config["client_secret"]).encode('utf-8')) headers = { "Authorization": "Basic " + header_token.decode('utf-8'), @@ -195,7 +197,7 @@ def refresh_credentials(self, config, config_path): post_body = { "grant_type": "refresh_token", - "refresh_token": config["refresh_token"], + "refresh_token": self.config["refresh_token"], } resp = self.session.post("https://identity.xero.com/connect/token", headers=headers, data=post_body) @@ -205,18 +207,19 @@ def refresh_credentials(self, config, config_path): resp = resp.json() # Write to config file - config['refresh_token'] = resp["refresh_token"] - update_config_file(config, config_path) + self.config['refresh_token'] = resp["refresh_token"] + self.config['access_token'] = resp["access_token"] + update_config_file(self.config, self.config_path) self.access_token = resp["access_token"] - self.tenant_id = config['tenant_id'] + self.tenant_id = self.config['tenant_id'] @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, @@ -232,7 +235,7 @@ def check_platform_access(self, config, config_path): if response.status_code != 200: raise_for_error(response) - + @backoff.on_exception(backoff.expo, XeroUnauthorizedError, max_tries=1) @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 filter(self, tap_stream_id, since=None, **params): @@ -249,6 +252,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 diff --git a/tap_xero/context.py b/tap_xero/context.py index fabf002..bf35bee 100644 --- a/tap_xero/context.py +++ b/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) diff --git a/tap_xero/streams.py b/tap_xero/streams.py index 051402c..28d5776 100644 --- a/tap_xero/streams.py +++ b/tap_xero/streams.py @@ -2,17 +2,18 @@ import singer from singer import metadata, metrics, Transformer from singer.utils import strptime_with_tz -import backoff from . import transform LOGGER = singer.get_logger() FULL_PAGE_SIZE = 100 -def _request_with_timer(tap_stream_id, xero, filter_options): + +def _make_request(ctx, tap_stream_id, filter_options=None): + filter_options = filter_options or {} with metrics.http_request_timer(tap_stream_id) as timer: try: - resp = xero.filter(tap_stream_id, **filter_options) + resp = ctx.client.filter(tap_stream_id, **filter_options) timer.tags[metrics.Tag.http_status_code] = 200 return resp except HTTPError as e: @@ -20,32 +21,6 @@ def _request_with_timer(tap_stream_id, xero, filter_options): raise -class RateLimitException(Exception): - pass - - -@backoff.on_exception(backoff.expo, - RateLimitException, - max_tries=10, - factor=2) -def _make_request(ctx, tap_stream_id, filter_options=None, attempts=0): - filter_options = filter_options or {} - try: - return _request_with_timer(tap_stream_id, ctx.client, filter_options) - except HTTPError as e: - if e.response.status_code == 401: - if attempts == 1: - raise Exception("Received Not Authorized response after credential refresh.") from e - ctx.refresh_credentials() - return _make_request(ctx, tap_stream_id, filter_options, attempts + 1) - - if e.response.status_code == 503: - raise RateLimitException() from e - - raise - assert False - - class Stream(): def __init__(self, tap_stream_id, pk_fields, bookmark_key="UpdatedDateUTC", format_fn=None): self.tap_stream_id = tap_stream_id diff --git a/tests/unittests/test_exception_handling.py b/tests/unittests/test_exception_handling.py index 74feadd..a270654 100644 --- a/tests/unittests/test_exception_handling.py +++ b/tests/unittests/test_exception_handling.py @@ -4,6 +4,7 @@ from unittest import mock import decimal import json +import io def mocked_session(*args, **kwargs): @@ -147,7 +148,8 @@ def test_json_decode_exception(self, mocked_session, mocked_jsondecode_failing_r config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: @@ -164,7 +166,8 @@ def test_normal_filter_execution(self, mocked_session, mocked_jsondecode_success config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: @@ -181,7 +184,8 @@ def test_badrequest_400_error(self, mocked_session, mocked_badrequest_400_error) config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -196,11 +200,13 @@ def test_badrequest_400_error(self, mocked_session, mocked_badrequest_400_error) @mock.patch('requests.Request', side_effect=mocked_unauthorized_401_error) - def test_unauthorized_401_error(self, mocked_session, mocked_unauthorized_401_error): + @mock.patch('tap_xero.XeroClient.refresh_credentials',) + def test_unauthorized_401_error(self, mocked_session, mocked_refresh_credentials, mocked_unauthorized_401_error): config = {} + tap_stream_id = "contacts" - - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -210,6 +216,7 @@ def test_unauthorized_401_error(self, mocked_session, mocked_unauthorized_401_er expected_error_message = "HTTP-error-code: 401, Error: Invalid authorization credentials." # Verifying the message formed for the custom exception + self.assertEqual(mocked_refresh_credentials.call_count, 3) self.assertEquals(str(e), expected_error_message) pass @@ -219,7 +226,8 @@ def test_forbidden_403_exception(self, mocked_session, mocked_forbidden_403_exce config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -238,7 +246,8 @@ def test_notfound_404_error(self, mocked_session, mocked_notfound_404_error): config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -256,7 +265,8 @@ def test_precondition_failed_412_error(self, mocked_session, mocked_precondition config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -274,7 +284,8 @@ def test_internalservererror_500_error(self, mocked_session, mocked_internalserv config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -293,7 +304,8 @@ def test_notimplemented_501_error(self, mocked_session, mocked_notimplemented_50 config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -311,7 +323,8 @@ def test_not_available_503_error(self, mocked_session, mocked_not_available_503_ config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -330,7 +343,8 @@ def test_too_many_requests_429_in_day_error(self, mocked_session, mocked_failed_ config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -350,7 +364,8 @@ def test_too_many_requests_429_in_minute_error(self, mocked_session, mocked_fail config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" @@ -370,7 +385,8 @@ def test_too_many_requests_in_day_429_not_backoff_behavior(self, mocked_session, config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: @@ -388,7 +404,8 @@ def test_too_many_requests_in_minute_429_backoff_behavior(self, mocked_session, config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: @@ -404,7 +421,8 @@ def test_internalservererror_500_backoff_behaviour(self, mocked_session, mocked_ config = {} tap_stream_id = "contacts" - xero_client = client_.XeroClient(config) + config_path = "" + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: @@ -429,11 +447,10 @@ def test_check_unauthorized_401_error_in_discovery_mode(self, mocked_unauthorize "tenant_id": "123" } config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroUnauthorizedError as e: expected_message = "HTTP-error-code: 401, Error: Invalid authorization credentials." self.assertEqual(str(e) ,expected_message) @@ -446,13 +463,12 @@ def test_check_forbidden_403_error_in_discovery_mode(self, mocked_refresh_creden mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroForbiddenError as e: expected_message = "HTTP-error-code: 403, Error: User doesn't have permission to access the resource." self.assertEqual(str(e) ,expected_message) @@ -465,13 +481,12 @@ def test_badrequest_400_error_in_discovery_mode(self, mocked_refresh_credentials mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroBadRequestError as e: expected_message = "HTTP-error-code: 400, Error: A validation exception has occurred." self.assertEqual(str(e), expected_message) @@ -484,13 +499,12 @@ def test_notfound_404_error_in_discovery_mode(self, mocked_refresh_credentials, mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroNotFoundError as e: expected_message = "HTTP-error-code: 404, Error: The resource you have specified cannot be found." self.assertEqual(str(e), expected_message) @@ -503,13 +517,12 @@ def test_precondition_failed_412_error_in_discovery_mode(self, mocked_refresh_cr mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroPreConditionFailedError as e: expected_message = "HTTP-error-code: 412, Error: One or more conditions given in the request header fields were invalid." self.assertEqual(str(e), expected_message) @@ -522,13 +535,12 @@ def test_internalservererror_500_error_in_discovery_mode(self, mocked_refresh_cr mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroInternalError as e: expected_message = "HTTP-error-code: 500, Error: An unhandled error with the Xero API. Contact the Xero API team if problems persist." self.assertEqual(str(e), expected_message) @@ -541,13 +553,12 @@ def test_notimplemented_501_error_in_discovery_mode(self, mocked_refresh_credent mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroNotImplementedError as e: expected_message = "HTTP-error-code: 501, Error: The method you have called has not been implemented." self.assertEqual(str(e), expected_message) @@ -560,13 +571,12 @@ def test_not_available_503_error_in_discovery_mode(self, mocked_refresh_credenti mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroNotAvailableError as e: expected_message = "HTTP-error-code: 503, Error: API service is currently unavailable." self.assertEqual(str(e), expected_message) @@ -579,13 +589,12 @@ def test_too_many_requests_in_day_429_error_in_discovery_mode(self, mocked_refre mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroTooManyError as e: expected_message = "HTTP-error-code: 429, Error: The API rate limit for your organisation/application pairing has been exceeded. Please retry after 1000 seconds" self.assertEqual(str(e), expected_message) @@ -598,13 +607,12 @@ def test_too_many_requests_in_minute_429_error_in_discovery_mode(self, mocked_re mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except client_.XeroTooManyInMinuteError as e: expected_message = "HTTP-error-code: 429, Error: The API rate limit for your organisation/application pairing has been exceeded. Please retry after 5 seconds" self.assertEqual(str(e), expected_message) @@ -617,13 +625,12 @@ def test_too_many_requests_in_day_429_not_backoff_behavior_discovery_mode(self, mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except (requests.HTTPError, client_.XeroTooManyError) as e: pass @@ -639,13 +646,12 @@ def test_too_many_requests_in_minute_429_backoff_behavior_discovery_mode(self, m mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except (requests.HTTPError, client_.XeroTooManyInMinuteError) as e: pass @@ -660,13 +666,12 @@ def test_internalservererror_500_backoff_behaviour_discovery_mode(self, mocked_r mocked_refresh_credentials.return_value = "" config = {} config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) xero_client.access_token = "123" xero_client.tenant_id = "123" try: - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() except (requests.HTTPError, client_.XeroInternalError) as e: pass @@ -688,12 +693,11 @@ def test_check_success_200_in_discovery_mode(self, mock_successful_session_post, "tenant_id": "123" } config_path = "" - - xero_client = client_.XeroClient(config) + xero_client = client_.XeroClient(config, config_path) expected_access_token = "123" expected_refresh_token = "345" - xero_client.check_platform_access(config, config_path) + xero_client.check_platform_access() self.assertEqual(xero_client.access_token, expected_access_token) self.assertEqual(config["refresh_token"], expected_refresh_token)