Add optional board filtering for Source Monday streams (#55713)

Co-authored-by: david.gold <[email protected]>
Co-authored-by: Natik Gadzhi <[email protected]>
Co-authored-by: Ian Alton <[email protected]>

Author: 4 people

airbyte-integrations/connectors/source-monday/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 80a54ea2-9959-4040-aac1-eee42423ec9b
-  dockerImageTag: 2.1.13
+  dockerImageTag: 2.2.0
   releases:
     breakingChanges:
       2.0.0:

airbyte-integrations/connectors/source-monday/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "2.1.13"
+version = "2.2.0"
 name = "source-monday"
 description = "Source implementation for Monday."
 authors = [ "Airbyte <[email protected]>",]

airbyte-integrations/connectors/source-monday/source_monday/graphql_requester.py

@@ -75,6 +75,13 @@ def _build_query(self, object_name: str, field_schema: dict, **object_arguments)
             else:
                 fields.append(field)
 
+        # when querying the boards stream (object_name == "boards"), filter by board_ids if they provided in the config
+        if object_name == "boards" and "board_ids" in self.config:
+            # if we are building a query for incremental syncs, board ids are already present under 'ids' key in object_arguments (as a result of fetching the activity_logs stream first)
+            # These ids are already an intersection of the board_ids provided in the config and the ones that must be fetched for the incremental sync and need not be overridden
+            if "ids" not in object_arguments:
+                object_arguments["ids"] = self.config.get("board_ids")
+
         arguments = self._get_object_arguments(**object_arguments)
         arguments = f"({arguments})" if arguments else ""
 
@@ -113,6 +120,9 @@ def _build_items_query(self, object_name: str, field_schema: dict, sub_page: Opt
             query = self._build_query("next_items_page", field_schema, limit=nested_limit, cursor=f'"{sub_page}"')
         else:
             query = self._build_query("items_page", field_schema, limit=nested_limit)
+            # since items are a subresource of boards, when querying items, filter by board_ids if provided in the config
+            if "board_ids" in self.config and "ids" not in object_arguments:
+                object_arguments["ids"] = self.config.get("board_ids")
             arguments = self._get_object_arguments(**object_arguments)
             query = f"boards({arguments}){{{query}}}"
 
@@ -156,7 +166,10 @@ def _build_activity_query(self, object_name: str, field_schema: dict, sub_page:
             created_at = datetime.fromtimestamp(created_at).strftime("%Y-%m-%dT%H:%M:%SZ")
 
         query = self._build_query(object_name, field_schema, limit=nested_limit, page=sub_page, fromt=created_at)
+        if "board_ids" in self.config and "ids" not in object_arguments:
+            object_arguments["ids"] = self.config.get("board_ids")
         arguments = self._get_object_arguments(**object_arguments)
+
         return f"boards({arguments}){{{query}}}"
 
     def get_request_headers(

airbyte-integrations/connectors/source-monday/source_monday/spec.json

@@ -72,6 +72,14 @@
             }
           }
         ]
+      },
+      "board_ids": {
+        "type": "array",
+        "title": "Boards to sync",
+        "description": "The IDs of the boards that the Items and Boards streams will extract records from. When left empty, streams will extract records from all boards that exist within the account.",
+        "items": {
+          "type": "integer"
+        }
       }
     }
   },

airbyte-integrations/connectors/source-monday/unit_tests/integrations/config.py

@@ -1,11 +1,12 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 
 class ConfigBuilder:
     def __init__(self) -> None:
         self._credentials: Dict[str, str] = {}
+        self._board_ids: List[int] = []
 
     def with_oauth_credentials(self, client_id: str, client_secret: str, access_token: str, subdomain: str) -> "ConfigBuilder":
         self._credentials["auth_type"] = "oauth2.0"
@@ -20,8 +21,14 @@ def with_api_token_credentials(self, api_token: str) -> "ConfigBuilder":
         self._credentials["auth_type"] = "api_token"
         return self
 
+    def with_board_ids(self, board_ids: List[int]) -> "ConfigBuilder":
+        self._board_ids = board_ids
+        return self
+
     def build(self) -> Dict[str, Any]:
         config = {}
         if self._credentials:
             config["credentials"] = self._credentials
+        if self._board_ids:
+            config["board_ids"] = self._board_ids
         return config

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_requests/__init__.py

@@ -1,3 +1,5 @@
 from .teams_requests_builder import TeamsRequestBuilder
+from .items_request_builder import ItemsRequestBuilder
+from .boards_request_builder import BoardsRequestBuilder
 
-__all__ = ["TeamsRequestBuilder"]
+__all__ = ["TeamsRequestBuilder", "ItemsRequestBuilder", "BoardsRequestBuilder"]

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_requests/base_requests_builder.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 
 import abc
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 from airbyte_cdk.test.mock_http import HttpRequest
 
@@ -55,3 +55,7 @@ def request_body(self):
     def with_authenticator(self, authenticator: Authenticator) -> "MondayBaseRequestBuilder":
         self._authenticator: Authenticator = authenticator
         return self
+
+    def with_board_ids(self, board_ids: List[int]) -> "MondayBaseRequestBuilder":
+        self._board_ids = board_ids
+        return self

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_requests/boards_request_builder.py

@@ -0,0 +1,27 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from typing import List
+
+from .base_requests_builder import MondayBaseRequestBuilder
+from .request_authenticators.authenticator import Authenticator
+
+
+class BoardsRequestBuilder(MondayBaseRequestBuilder):
+    @classmethod
+    def boards_endpoint(cls, authenticator: Authenticator, board_ids: List[int] = None) -> "BoardsRequestBuilder":
+        return cls().with_authenticator(authenticator).with_board_ids(board_ids)
+
+    @property
+    def query_params(self):
+        params = super().query_params or {}
+        if self._board_ids:
+            board_ids = ", ".join(list(map(str, self._board_ids)))
+            board_ids_str = f",ids:[{board_ids}]"
+        else:
+            board_ids_str = ""
+
+        params["query"] = (
+            "query{boards(limit:10%s){id,name,board_kind,type,columns{archived,description,id,settings_str,title,type,width},communication,description,groups{archived,color,deleted,id,position,title},owners{id},creator{id},permissions,state,subscribers{id},tags{id},top_group{id},updated_at,updates{id},views{id,name,settings_str,type,view_specific_data_str},workspace{id,name,kind,description}}}"
+            % board_ids_str
+        )
+        return params

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_requests/items_request_builder.py

@@ -0,0 +1,27 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from typing import List
+
+from .base_requests_builder import MondayBaseRequestBuilder
+from .request_authenticators.authenticator import Authenticator
+
+
+class ItemsRequestBuilder(MondayBaseRequestBuilder):
+    @classmethod
+    def items_endpoint(cls, authenticator: Authenticator, board_ids: List[int] = None) -> "ItemsRequestBuilder":
+        return cls().with_authenticator(authenticator).with_board_ids(board_ids)
+
+    @property
+    def query_params(self):
+        params = super().query_params or {}
+        if self._board_ids:
+            board_ids = ", ".join(list(map(str, self._board_ids)))
+            board_ids_str = f",ids:[{board_ids}]"
+        else:
+            board_ids_str = ""
+
+        params["query"] = (
+            "query{boards(limit:1%s){items_page(limit:20){cursor,items{id,name,assets{created_at,file_extension,file_size,id,name,original_geometry,public_url,uploaded_by{id},url,url_thumbnail},board{id,name},column_values{id,text,type,value,... on MirrorValue{display_value},... on BoardRelationValue{display_value},... on DependencyValue{display_value}},created_at,creator_id,group{id},parent_item{id},state,subscribers{id},updated_at,updates{id}}}}}"
+            % board_ids_str
+        )
+        return params

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/__init__.py

@@ -1,4 +1,6 @@
 from .error_response_builder import ErrorResponseBuilder
 from .teams_response_builder import TeamsResponseBuilder
+from .items_response_builder import ItemsResponseBuilder
+from .boards_response_builder import BoardsResponseBuilder
 
-__all__ = ["ErrorResponseBuilder", "TeamsResponseBuilder"]
+__all__ = ["ErrorResponseBuilder", "TeamsResponseBuilder", "ItemsResponseBuilder", "BoardsResponseBuilder"]

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/boards_response_builder.py

@@ -0,0 +1,9 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.test.mock_http.response_builder import HttpResponseBuilder, NestedPath, find_template
+
+
+class BoardsResponseBuilder(HttpResponseBuilder):
+    @classmethod
+    def boards_response(cls) -> "BoardsResponseBuilder":
+        return cls(find_template("boards", __file__), NestedPath(["data", "boards"]), None)

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/items_response_builder.py

@@ -0,0 +1,9 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.test.mock_http.response_builder import HttpResponseBuilder, NestedPath, find_template
+
+
+class ItemsResponseBuilder(HttpResponseBuilder):
+    @classmethod
+    def items_response(cls) -> "ItemsResponseBuilder":
+        return cls(find_template("items", __file__), NestedPath(["data", "boards", 0, "items_page", "items"]), None)

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/records/__init__.py

@@ -1,3 +1,5 @@
 from .teams_record_builder import TeamsRecordBuilder
+from .items_record_builder import ItemsRecordBuilder
+from .boards_record_builder import BoardsRecordBuilder
 
-__all__ = ["TeamsRecordBuilder"]
+__all__ = ["TeamsRecordBuilder", "ItemsRecordBuilder", "BoardsRecordBuilder"]

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/records/boards_record_builder.py

@@ -0,0 +1,12 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.test.mock_http.response_builder import FieldPath, NestedPath
+
+from .record_builder import MondayRecordBuilder
+
+
+class BoardsRecordBuilder(MondayRecordBuilder):
+    @classmethod
+    def boards_record(cls) -> "BoardsRecordBuilder":
+        record_template = cls.extract_record("boards", __file__, NestedPath(["data", "boards", 0]))
+        return cls(record_template, FieldPath("id"), None)

airbyte-integrations/connectors/source-monday/unit_tests/integrations/monday_responses/records/items_record_builder.py

@@ -0,0 +1,12 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.test.mock_http.response_builder import FieldPath, NestedPath
+
+from .record_builder import MondayRecordBuilder
+
+
+class ItemsRecordBuilder(MondayRecordBuilder):
+    @classmethod
+    def items_record(cls) -> "ItemsRecordBuilder":
+        record_template = cls.extract_record("items", __file__, NestedPath(["data", "boards", 0, "items_page", "items", 0]))
+        return cls(record_template, FieldPath("id"), None)

airbyte-integrations/connectors/source-monday/unit_tests/integrations/test_boards_stream.py

@@ -0,0 +1,53 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from unittest import TestCase
+
+from airbyte_cdk.test.mock_http import HttpMocker
+from airbyte_protocol.models import SyncMode
+
+from .config import ConfigBuilder
+from .monday_requests import BoardsRequestBuilder
+from .monday_requests.request_authenticators import ApiTokenAuthenticator
+from .monday_responses import BoardsResponseBuilder
+from .monday_responses.records import BoardsRecordBuilder
+from .utils import read_stream
+
+
+class TestBoardsStreamFullRefresh(TestCase):
+    def get_authenticator(self, config):
+        return ApiTokenAuthenticator(api_token=config["credentials"]["api_token"])
+
+    @HttpMocker()
+    def test_read_board_record(self, http_mocker):
+        """
+        A full refresh sync of the boards stream without pagination or board filtering
+        """
+        config = ConfigBuilder().with_api_token_credentials("api-token").build()
+        api_token_authenticator = self.get_authenticator(config)
+
+        http_mocker.get(
+            BoardsRequestBuilder.boards_endpoint(api_token_authenticator).build(),
+            BoardsResponseBuilder.boards_response()
+            .with_record(BoardsRecordBuilder.boards_record())
+            .with_record(BoardsRecordBuilder.boards_record())
+            .build(),
+        )
+
+        output = read_stream("boards", SyncMode.full_refresh, config)
+        assert len(output.records) == 2
+
+    @HttpMocker()
+    def test_read_with_board_ids_filter(self, http_mocker):
+        """
+        A full refresh sync of the boards stream with board filtering and without pagination
+        """
+        board_ids = [123, 456]
+        config = ConfigBuilder().with_api_token_credentials("api-token").with_board_ids(board_ids).build()
+        api_token_authenticator = self.get_authenticator(config)
+
+        http_mocker.get(
+            BoardsRequestBuilder.boards_endpoint(api_token_authenticator, board_ids).build(),
+            BoardsResponseBuilder.boards_response().with_record(BoardsRecordBuilder.boards_record()).build(),
+        )
+
+        output = read_stream("boards", SyncMode.full_refresh, config)
+        assert len(output.records) == 1

airbyte-integrations/connectors/source-monday/unit_tests/integrations/test_items_stream.py

@@ -0,0 +1,53 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from unittest import TestCase
+
+from airbyte_cdk.test.mock_http import HttpMocker
+from airbyte_protocol.models import SyncMode
+
+from .config import ConfigBuilder
+from .monday_requests import ItemsRequestBuilder
+from .monday_requests.request_authenticators import ApiTokenAuthenticator
+from .monday_responses import ItemsResponseBuilder
+from .monday_responses.records import ItemsRecordBuilder
+from .utils import read_stream
+
+
+class TestItemsStreamFullRefresh(TestCase):
+    def get_authenticator(self, config):
+        return ApiTokenAuthenticator(api_token=config["credentials"]["api_token"])
+
+    @HttpMocker()
+    def test_read_item_record(self, http_mocker):
+        """
+        A full refresh sync of the items stream without pagination or board filtering
+        """
+        config = ConfigBuilder().with_api_token_credentials("api-token").build()
+        api_token_authenticator = self.get_authenticator(config)
+
+        http_mocker.get(
+            ItemsRequestBuilder.items_endpoint(api_token_authenticator).build(),
+            ItemsResponseBuilder.items_response()
+            .with_record(ItemsRecordBuilder.items_record())
+            .with_record(ItemsRecordBuilder.items_record())
+            .build(),
+        )
+
+        output = read_stream("items", SyncMode.full_refresh, config)
+        assert len(output.records) == 2
+
+    @HttpMocker()
+    def test_read_with_board_ids_filter(self, http_mocker):
+        """
+        A full refresh sync of the items stream with board filtering and without pagination
+        """
+        board_ids = [123, 456]
+        config = ConfigBuilder().with_api_token_credentials("api-token").with_board_ids(board_ids).build()
+        api_token_authenticator = self.get_authenticator(config)
+
+        http_mocker.get(
+            ItemsRequestBuilder.items_endpoint(api_token_authenticator, board_ids).build(),
+            ItemsResponseBuilder.items_response().with_record(ItemsRecordBuilder.items_record()).build(),
+        )
+
+        output = read_stream("items", SyncMode.full_refresh, config)
+        assert len(output.records) == 1