Added implementation to exclude unauthorized streams from discovery

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+# 1.2.0
+  * Streams returning 403 Forbidden during discovery are now excluded from the catalog; discovery fails only if none are accessible. [#52](https://github.com/singer-io/tap-gitlab/pull/52)
+
 # 1.1.2
   * `projects` now independently fetches, merges, and deduplicates project IDs from configured groups. [#55](https://github.com/singer-io/tap-gitlab/pull/55)
 

setup.py

@@ -5,7 +5,7 @@
 
 setup(
     name='tap-gitlab',
-    version='1.1.2',
+    version='1.2.0',
     description='Singer.io tap for extracting data from the GitLab API',
     author='Stitch',
     url='https://singer.io',

tap_gitlab/__init__.py

@@ -11,12 +11,12 @@
 REQUIRED_CONFIG_KEYS = ["private_token", "start_date", "groups", "projects"]
 
 
-def do_discover():
+def do_discover(client):
     """
     Discover and emit the catalog to stdout
     """
     LOGGER.info("Starting discover")
-    catalog = discover()
+    catalog = discover(client)
     json.dump(catalog.to_dict(), sys.stdout, indent=2)
     LOGGER.info("Finished discover")
 
@@ -33,7 +33,7 @@ def main():
 
     with Client(parsed_args.config) as client:
         if parsed_args.discover:
-            do_discover()
+            do_discover(client)
         elif parsed_args.catalog:
             sync(
                 client=client,

tap_gitlab/discover.py

@@ -2,15 +2,70 @@
 from singer import metadata
 from singer.catalog import Catalog, CatalogEntry, Schema
 from tap_gitlab.schema import get_schemas
+from tap_gitlab.streams import STREAMS
+from tap_gitlab.exceptions import ForbiddenError
 
 LOGGER = singer.get_logger()
 
 
-def discover() -> Catalog:
+def _prune_inaccessible_children(schemas: dict, field_metadata: dict) -> None:
+    """
+    Remove child streams from the catalog whose parent stream was excluded.
+    Mutates schemas and field_metadata in place.
+    """
+    to_remove = []
+    for name, stream_cls in list(STREAMS.items()):
+        parent = getattr(stream_cls, "parent", None)
+        if name in schemas and parent and parent not in schemas:
+            LOGGER.warning(
+                "Stream '%s' excluded from catalog because its parent stream '%s' is not accessible.",
+                name, parent,
+            )
+            schemas.pop(name, None)
+            field_metadata.pop(name, None)
+            to_remove.append(name)
+    return to_remove
+
+
+def _apply_access_checks(client, schemas: dict, field_metadata: dict) -> None:
+    """
+    Probe each stream for read access and remove inaccessible streams
+    (and their children) from schemas and field_metadata in place.
+    Raises ForbiddenError if no parent streams are accessible.
+    """
+    inaccessible_streams = [
+        stream_name
+        for stream_name, stream_cls in STREAMS.items()
+        if stream_name in schemas
+        and not stream_cls(client=client).check_access()
+    ]
+
+    for stream_name in inaccessible_streams:
+        schemas.pop(stream_name, None)
+        field_metadata.pop(stream_name, None)
+
+    inaccessible_streams.extend(_prune_inaccessible_children(schemas, field_metadata))
+
+    if not schemas:
+        raise ForbiddenError(
+            "No streams are accessible. Ensure the credentials have read permission for at least one stream."
+        )
+    elif inaccessible_streams:
+        LOGGER.warning(
+            "Unauthorized streams excluded from catalog: %s",
+            ", ".join(sorted(set(inaccessible_streams))),
+        )
+
+
+def discover(client) -> Catalog:
     """
     Run the discovery mode, prepare the catalog file and return the catalog.
+    Access to each stream is verified using the provided client and streams
+    the credentials cannot read are excluded from the returned catalog.
     """
     schemas, field_metadata = get_schemas()
+    _apply_access_checks(client, schemas, field_metadata)
+
     catalog = Catalog([])
 
     for stream_name, schema_dict in schemas.items():

tap_gitlab/schemas/branches.json

@@ -7,6 +7,13 @@
                 "integer"
             ]
         },
+        "projects_updated_at": {
+            "type": [
+                "null",
+                "string"
+            ],
+            "format": "date-time"
+        },
         "name": {
             "type": [
                 "null",

tap_gitlab/schemas/users.json

@@ -13,6 +13,13 @@
                 "integer"
             ]
         },
+        "projects_updated_at": {
+            "type": [
+                "null",
+                "string"
+            ],
+            "format": "date-time"
+        },
         "username": {
             "type": [
                 "null",

tap_gitlab/streams/abstracts.py

@@ -13,7 +13,10 @@
 from dateutil import parser
 from datetime import datetime, timezone
 
+from tap_gitlab.exceptions import ForbiddenError, UnauthorizedError
+
 LOGGER = get_logger()
+LOCAL_TIMEZONE = datetime.now().astimezone().tzinfo
 
 
 class BaseStream(ABC):
@@ -31,8 +34,8 @@ class BaseStream(ABC):
     def __init__(self, client=None, catalog=None) -> None:
         self.client = client
         self.catalog = catalog
-        self.schema = self.catalog.schema.to_dict() if self.catalog else None
-        self.metadata = metadata.to_map(self.catalog.metadata) if self.catalog else None
+        self.schema = self.catalog.schema.to_dict() if self.catalog else {}
+        self.metadata = metadata.to_map(self.catalog.metadata) if self.catalog else {}
         self.child_to_sync = []
         self.params = {}
 
@@ -116,8 +119,33 @@ def modify_object(self, record: Dict, parent_record: Dict = None) -> Dict:
     def get_url_endpoint(self, parent_obj: Dict = None) -> str:
         return self.url_endpoint or f"{self.client.base_url}/{self.path}"
 
+    def check_access(self) -> bool:
+        """
+        Verify that the API credentials have read access to this stream.
+        Returns True if accessible, False if a 403 Forbidden error is raised.
+        Child streams always return True (access is governed by the parent check).
+        """
+        if self.parent:
+            return True
+
+        url = self.get_url_endpoint()
+        params = {"per_page": 1}
+
+        try:
+            self.client.get(url, params, self.headers, None)
+            return True
+        except (ForbiddenError, UnauthorizedError) as exc:
+            LOGGER.warning(
+                "Unauthorized Stream: %s, excluding from catalog. HTTP-Error-Message:'%s'",
+                self.tap_stream_id,
+                str(exc),
+            )
+            return False
+
 
 class IncrementalStream(BaseStream):
+    send_updated_since = True
+
     def get_bookmark(self, state: dict, stream: str, key: Any = None) -> int:
         return get_bookmark(  # pylint: disable=E1121
             state,
@@ -135,7 +163,15 @@ def update_bookmark_state(self, state: dict, stream: str, key: Any = None, value
             state, stream, bookmark_key, self.client.config["start_date"]
         )
         try:
-            value = max(current_bookmark, value)
+            current_dt = self._to_utc_datetime(current_bookmark)
+            value_dt = self._to_utc_datetime(value)
+
+            if current_dt and value_dt:
+                value = max(current_dt, value_dt).isoformat(timespec='seconds').replace('+00:00', 'Z')
+            elif value_dt:
+                value = value_dt.isoformat(timespec='seconds').replace('+00:00', 'Z')
+            else:
+                value = current_bookmark
         except Exception:
             LOGGER.warning("Failed to compare bookmark values. Keeping current bookmark.")
             value = current_bookmark
@@ -153,14 +189,14 @@ def _to_utc_datetime(self, value):
             return None
         if isinstance(value, datetime):
             if value.tzinfo is None:
-                return value.replace(tzinfo=timezone.utc)
+                return value.replace(tzinfo=LOCAL_TIMEZONE).astimezone(timezone.utc)
             return value.astimezone(timezone.utc)
         if isinstance(value, (int, float)):
-            return datetime.fromtimestamp(value).replace(tzinfo=timezone.utc)
+            return datetime.fromtimestamp(value, tz=timezone.utc)
         if isinstance(value, str):
             dt = parser.parse(value)
             if dt.tzinfo is None:
-                dt = dt.replace(tzinfo=timezone.utc)
+                dt = dt.replace(tzinfo=LOCAL_TIMEZONE)
             return dt.astimezone(timezone.utc)
         LOGGER.warning(f"Unsupported timestamp type: {type(value)}")
         return None
@@ -178,7 +214,8 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) -
             bookmark_date = self._to_utc_datetime(self.client.config["start_date"])
 
         current_max_bookmark_date = bookmark_date
-        self.update_params(updated_since=bookmark_date.isoformat(timespec='seconds').replace('+00:00', 'Z'))
+        if self.send_updated_since:
+            self.update_params(updated_since=bookmark_date.isoformat(timespec='seconds').replace('+00:00', 'Z'))
         self.url_endpoint = self.get_url_endpoint(parent_obj)
 
         with metrics.record_counter(self.tap_stream_id) as counter:
@@ -194,6 +231,12 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) -
                     LOGGER.warning(f"Skipping record with invalid {self.replication_keys[0]}: {record_value}")
                     continue
 
+                # Normalize replication key values to UTC with explicit timezone
+                # so emitted records and saved state are compared consistently.
+                transformed_record[self.replication_keys[0]] = (
+                    record_timestamp.isoformat(timespec='microseconds').replace('+00:00', 'Z')
+                )
+
                 if record_timestamp >= bookmark_date:
                     if self.is_selected():
                         write_record(self.tap_stream_id, transformed_record)
@@ -215,6 +258,34 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) -
             )
             return counter.value
 
+
+class ParentBaseStream(IncrementalStream):
+    """Incremental parent stream that owns child stream bookmarks."""
+
+    def get_bookmark(self, state: dict, stream: str, key: Any = None):
+        min_bookmark = super().get_bookmark(state, stream) if self.is_selected() else None
+        bookmark_key = f"{self.tap_stream_id}_{self.replication_keys[0]}"
+
+        for child in self.child_to_sync:
+            child_bookmark = super().get_bookmark(
+                state, child.tap_stream_id, key=bookmark_key
+            )
+            min_bookmark = min(min_bookmark, child_bookmark) if min_bookmark else child_bookmark
+
+        return min_bookmark or self.client.config["start_date"]
+
+    def update_bookmark_state(self, state: dict, stream: str, key: Any = None, value: Any = None) -> Dict:
+        if self.is_selected():
+            super().update_bookmark_state(state, stream, key=key, value=value)
+
+        bookmark_key = f"{self.tap_stream_id}_{self.replication_keys[0]}"
+        for child in self.child_to_sync:
+            super().update_bookmark_state(
+                state, child.tap_stream_id, key=bookmark_key, value=value
+            )
+
+        return state
+
 class FullTableStream(BaseStream):
     """Base Class for FullTable Stream."""
 
@@ -245,6 +316,8 @@ def sync(
 
 class ChildBaseStream(IncrementalStream):
     """Base Class for Child Stream."""
+    send_updated_since = False
+
     def get_bookmark(self, state: Dict, stream: str, key: Any = None) -> int:
         """Singleton bookmark value for child streams."""
         if not self.bookmark_value:

tap_gitlab/streams/branches.py

@@ -1,14 +1,14 @@
 from typing import Dict, Any
 from urllib.parse import quote
-from tap_gitlab.streams.abstracts import FullTableStream
+from tap_gitlab.streams.abstracts import ChildBaseStream
 
 
-class Branches(FullTableStream):
+class Branches(ChildBaseStream):
     tap_stream_id = "branches"
     key_properties = ["project_id", "name"]
-    replication_method = "FULL_TABLE"
+    replication_method = "INCREMENTAL"
     parent = "projects"
-    replication_keys = None
+    replication_keys = ["projects_updated_at"]
     path = "projects/{}/repository/branches"
     data_key = None
 
@@ -30,9 +30,9 @@ def get_url_endpoint(self, parent_obj: Dict = None) -> str:
         return endpoint
 
     def modify_object(self, record, parent_record=None):
-        """Adding project_id and last_committed_date to the record"""
-        if isinstance(record, dict):
-            if parent_record and isinstance(parent_record, dict):
-                record["project_id"] = parent_record.get("id")
+        """Add project_id and the parent project's update timestamp."""
+        if isinstance(record, dict) and isinstance(parent_record, dict):
+            record["project_id"] = parent_record.get("id")
+            record["projects_updated_at"] = parent_record.get("updated_at")
 
         return record

tap_gitlab/streams/projects.py

@@ -2,11 +2,11 @@
 from singer import get_logger
 from urllib.parse import quote
 
-from tap_gitlab.streams.abstracts import IncrementalStream
+from tap_gitlab.streams.abstracts import ParentBaseStream
 
 LOGGER = get_logger()
 
-class Projects(IncrementalStream):
+class Projects(ParentBaseStream):
     tap_stream_id = "projects"
     key_properties = ["id"]
     replication_method = "INCREMENTAL"

tap_gitlab/streams/users.py

@@ -1,14 +1,14 @@
 from typing import Dict, Any
 from urllib.parse import quote
-from tap_gitlab.streams.abstracts import FullTableStream
+from tap_gitlab.streams.abstracts import ChildBaseStream
 
 
-class Users(FullTableStream):
+class Users(ChildBaseStream):
     tap_stream_id = "users"
     key_properties = ["id", "project_id"]
-    replication_method = "FULL_TABLE"
+    replication_method = "INCREMENTAL"
     parent = "projects"
-    replication_keys = None
+    replication_keys = ["projects_updated_at"]
     path = "projects/{}/users"
     data_key = None
 
@@ -29,9 +29,10 @@ def get_url_endpoint(self, parent_obj: Dict = None) -> str:
         endpoint = f"{self.client.base_url}/{self.get_url(parent_obj)}"
         return endpoint
 
-    def modify_object(self, record, parent_record = None):
-        """Adding project_id to the record."""
-        if isinstance(record, dict):
+    def modify_object(self, record, parent_record=None):
+        """Add project_id and the parent project's update timestamp."""
+        if isinstance(record, dict) and isinstance(parent_record, dict):
             record["project_id"] = parent_record.get("id")
+            record["projects_updated_at"] = parent_record.get("updated_at")
 
         return record