SAC-30896: Handle un-auth stream exclusion from catalog

Author: satyendra101

setup.py

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

tap_xero/__init__.py

@@ -5,7 +5,7 @@
 from singer import metadata, metrics, utils
 from singer.catalog import Catalog, CatalogEntry, Schema
 from . import streams as streams_
-from .client import XeroClient
+from .client import XeroClient, XeroForbiddenError
 from .context import Context
 
 REQUIRED_CONFIG_KEYS = [
@@ -67,10 +67,15 @@ def load_metadata(stream, schema):
 def ensure_credentials_are_valid(config):
     XeroClient(config).filter("currencies")
 
+
 def discover(ctx):
+    LOGGER.info("Starting discover")
     ctx.check_platform_access()
     catalog = Catalog([])
     for stream in streams_.all_streams:
+        if not stream.check_access(ctx):
+            continue
+
         schema_dict = load_schema(stream.tap_stream_id)
         mdata = load_metadata(stream, schema_dict)
 
@@ -82,6 +87,13 @@ def discover(ctx):
             schema=schema,
             metadata=mdata
         ))
+
+    if not catalog.streams:
+        error_msg = "The provided credentials doesn't have access to any streams. Please recheck configuration"
+        LOGGER.error(error_msg)
+        raise XeroForbiddenError(error_msg)
+
+    LOGGER.info("Finished discover")
     return catalog
 
 
@@ -126,6 +138,8 @@ def main_impl():
 
         sync(Context(args.config, args.state, catalog, args.config_path))
 
+
+@singer.utils.handle_top_exception(LOGGER)
 def main():
     try:
         main_impl()

tap_xero/streams.py

@@ -1,9 +1,13 @@
-from requests.exceptions import HTTPError
+import backoff
 import singer
-from singer import metadata, metrics, Transformer
+from requests.exceptions import HTTPError
+from singer import Transformer, metadata, metrics
 from singer.utils import strptime_with_tz
-import backoff
+
+from tap_xero.context import Context
+
 from . import transform
+from .client import XeroForbiddenError, XeroUnauthorizedError
 
 LOGGER = singer.get_logger()
 FULL_PAGE_SIZE = 100
@@ -54,6 +58,28 @@ def __init__(self, tap_stream_id, pk_fields, bookmark_key="UpdatedDateUTC", form
         self.bookmark_key = bookmark_key
         self.replication_method = "INCREMENTAL"
         self.filter_options = {}
+        self.probe_filter_options = {}
+
+    def check_access(self, ctx: Context):
+        """
+        Verify that the API credentials have read access to this stream.
+        Returns True if accessible, False if a 403 Forbidden or 401 Unauthorized error is raised.
+        """
+        ctx.refresh_credentials()
+
+        try:
+            LOGGER.info("Checking access for stream %s...", self.tap_stream_id)
+            ctx.client.filter(self.tap_stream_id, self.probe_filter_options)
+            LOGGER.info("Stream %s is accessible with the provided credentials.", self.tap_stream_id)
+
+            return True
+        except (XeroUnauthorizedError, XeroForbiddenError):
+            LOGGER.warning(
+                "Stream '%s' does not have read permission, "
+                "excluding from catalog.",
+                self.tap_stream_id,
+            )
+            return False
 
     def metrics(self, records):
         with metrics.record_counter(self.tap_stream_id) as counter:
@@ -135,6 +161,8 @@ class Journals(Stream):
     """The Journals endpoint is a special case. It has its own way of ordering
     and paging the data. See
     https://developer.xero.com/documentation/api/journals"""
+    probe_filter_options = {"offset": 0}
+
     def sync(self, ctx):
         bookmark = [self.tap_stream_id, self.bookmark_key]
         journal_number = ctx.get_bookmark(bookmark) or 0
@@ -158,6 +186,8 @@ class LinkedTransactions(Stream):
     the UpdatedDateUTC timestamp in them. Therefore we must always iterate over
     all of the data, but we can manually omit records based on the
     UpdatedDateUTC property."""
+    probe_filter_options = {"page": 1}
+
     def sync(self, ctx):
         bookmark = [self.tap_stream_id, self.bookmark_key]
         offset = [self.tap_stream_id, "page"]