source-monday: refactors board items fetching and cursor handling

Simplifies the logic for fetching items from boards by removing the iterator class, consolidating the process into standalone async functions, and clarifying error handling for invalid inputs.

Fixes a bug in cursor collection to support both board-level and next-page cursors, enhancing reliability and maintainability of pagination.

Removes unused logic related to tracking the oldest updated board, focusing on streamlined item retrieval.

Author: JustinASmith

source-monday/source_monday/api.py

@@ -12,7 +12,7 @@
     fetch_boards_minimal,
     fetch_boards_paginated,
     fetch_items_by_id,
-    BoardItemIterator,
+    get_items_from_boards,
 )
 from source_monday.models import (
     FullRefreshResource,
@@ -273,12 +273,12 @@ async def fetch_items_changes(
             },
         )
 
-        item_iterator = BoardItemIterator(http, log)
-
         while True:
-            items_generator, get_board_cursor = await item_iterator.get_items_from_boards(list(board_ids_to_backfill))
-
-            async for item in items_generator:
+            async for item in get_items_from_boards(
+                http,
+                log,
+                list(board_ids_to_backfill),
+            ):
                 if window_start <= item.updated_at <= window_end:
                     if item.state == "deleted":
                         item.meta_ = Item.Meta(op="d")
@@ -289,18 +289,6 @@ async def fetch_items_changes(
                     docs_emitted += 1
                     yield item
 
-            board_cursor = get_board_cursor()
-
-            if not board_cursor:
-                log.debug(
-                    f"No more items to fetch for {len(board_ids_to_backfill)} boards",
-                    {
-                        "board_ids": list(board_ids_to_backfill),
-                        "board_cursor": board_cursor,
-                    },
-                )
-                break
-
     if docs_emitted > 0:
         yield max_updated_at_in_window + timedelta(seconds=1)
     else:

source-monday/source_monday/graphql/__init__.py

@@ -12,7 +12,7 @@
     TEAMS,
     USERS,
 )
-from .items import fetch_items_by_id, BoardItemIterator
+from .items import fetch_items_by_id, get_items_from_boards
 from .query_executor import execute_query
 
 __all__ = [
@@ -28,5 +28,5 @@
     "fetch_boards_paginated",
     "fetch_boards_with_retry",
     "fetch_items_by_id",
-    "BoardItemIterator",
+    "get_items_from_boards",
 ]

source-monday/source_monday/graphql/boards.py

@@ -51,19 +51,18 @@ async def fetch_boards_minimal(
     log: Logger,
 ) -> AsyncGenerator[Board, None]:
     query = """
-    query ($limit: Int = 500, $page: Int = 1, $state: State = all) {
+    query ($limit: Int = 10000, $page: Int = 1, $state: State = all) {
         boards(limit: $limit, page: $page, state: $state, order_by: created_at) {
             id
             name
             state
             updated_at
-            items_count
         }
     }
     """
 
     page = 1
-    limit = 500
+    limit = 10000
 
     while True:
         variables = {