perf(notion_datasource): speed up get_authorized_pages for large workspaces

- Parallelize parent block resolution with ThreadPoolExecutor (configurable via NOTION_PARENT_RESOLVE_WORKERS, default 8) and memoize lookups with a thread-safe cache so shared ancestors are not refetched.
- Replace the two filtered /v1/search loops (one for pages, one for databases) with a single un-filtered pass dispatched by object type, halving the search round-trips.
- Route the three direct requests.* call sites through _make_request so that 429 / transient 5xx are retried uniformly. _make_request now accepts allow_status to preserve the existing 404 -> root fallback for inaccessible parent blocks.

Public method signatures (notion_page_search, notion_database_search, notion_block_parent_page_id) are preserved as thin wrappers so existing callers keep working.

Bump plugin version to 0.1.19.

Author: Kota-Maeda

datasources/notion_datasource/datasources/utils/notion_client.py

@@ -3,14 +3,20 @@
 This module provides a unified interface for interacting with the Notion API
 """
 
+import os
+import threading
 import time
+from collections.abc import Iterable
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 import requests
 
 from dify_plugin.entities.datasource import OnlineDocumentPage
 
 __TIMEOUT_SECONDS__ = 60 * 10
+_DEFAULT_PARENT_RESOLVE_WORKERS = 8
+_MAX_PARENT_RESOLVE_WORKERS = 32
 
 
 class NotionClient:
@@ -40,6 +46,10 @@ def __init__(self, integration_token: str):
             "Notion-Version": self._API_VERSION,
             "Content-Type": "application/json",
         }
+        # Memoize parent_id lookups so the threaded resolution in
+        # get_authorized_pages doesn't re-fetch the same ancestor block twice.
+        self._parent_cache: dict[str, str] = {}
+        self._parent_cache_lock = threading.Lock()
 
     def _make_request(
         self,
@@ -48,7 +58,8 @@ def _make_request(
         params: dict[str, Any] | None = None,
         json_data: dict[str, Any] | None = None,
         max_retries: int = 3,
-    ) -> dict[str, Any]:
+        allow_status: Iterable[int] = (),
+    ) -> dict[str, Any] | None:
         """
         Make an API request to Notion with retry logic for rate limits.
 
@@ -58,12 +69,15 @@ def _make_request(
             params: URL parameters for GET requests
             json_data: JSON data for POST/PATCH requests
             max_retries: Maximum number of retries for rate limiting
+            allow_status: HTTP status codes that should resolve to None
+                instead of raising (e.g. {404} for optional resources).
 
         Returns:
-            Response data as dictionary
+            Response data as dictionary, or None if the response status was in allow_status.
         """
         url = f"{self._API_BASE_URL}{endpoint}"
         retries = 0
+        allow_status_set = set(allow_status)
 
         while retries <= max_retries:
             try:
@@ -83,6 +97,9 @@ def _make_request(
                     retries += 1
                     continue
 
+                if response.status_code in allow_status_set:
+                    return None
+
                 response.raise_for_status()
                 return response.json()
 
@@ -439,145 +456,126 @@ def format_rich_text(self, content: str) -> list[dict[str, Any]]:
         """
         return self.create_rich_text(content)
 
-    def get_authorized_pages(self):
-        pages = []
-        access_token = self.integration_token
-        page_results = self.notion_page_search(access_token)
-        database_results = self.notion_database_search(access_token)
-        # get page detail
-        for page_result in page_results:
-            page_id = page_result["id"]
-            page_name = "Untitled"
-            for key in page_result["properties"]:
-                if "title" in page_result["properties"][key] and page_result["properties"][key]["title"]:
-                    title_list = page_result["properties"][key]["title"]
-                    if len(title_list) > 0 and "plain_text" in title_list[0]:
-                        page_name = title_list[0]["plain_text"]
-            icon = None
-            parent = page_result["parent"]
-            parent_type = parent["type"]
-            try:
-                if parent_type == "block_id":
-                    parent_id = self.notion_block_parent_page_id(access_token, parent[parent_type])
-                elif parent_type == "workspace":
-                    parent_id = "root"
-                else:
-                    parent_id = parent[parent_type]
-            except ValueError:
-                continue
-            page = OnlineDocumentPage(
-                page_id=page_id,
-                page_name=page_name,
-                page_icon=icon,
-                parent_id=parent_id,
-                type="page",
-                last_edited_time=page_result["last_edited_time"],
-            )
-            pages.append(page)
-            # get database detail
-        for database_result in database_results:
-            page_id = database_result["id"]
-            page_name = database_result["title"][0]["plain_text"] if len(database_result["title"]) > 0 else "Untitled"
-
-            icon = None
-            parent = database_result["parent"]
-            parent_type = parent["type"]
-            try:
-                if parent_type == "block_id":
-                    parent_id = self.notion_block_parent_page_id(access_token, parent[parent_type])
-                elif parent_type == "workspace":
-                    parent_id = "root"
-                else:
-                    parent_id = parent[parent_type]
-            except ValueError:
-                continue
-            page = OnlineDocumentPage(
-                page_id=page_id,
-                page_name=page_name,
-                page_icon=icon,
-                parent_id=parent_id,
-                type="database",
-                last_edited_time=database_result["last_edited_time"],
-            )
-            pages.append(page)
+    def get_authorized_pages(self) -> list[OnlineDocumentPage]:
+        items = self._search_all()
+        worker_count = self._resolve_worker_count()
+
+        pages: list[OnlineDocumentPage] = []
+        if worker_count <= 1:
+            for item in items:
+                entry = self._build_page_entry(item)
+                if entry is not None:
+                    pages.append(entry)
+            return pages
+
+        with ThreadPoolExecutor(max_workers=worker_count) as executor:
+            for entry in executor.map(self._build_page_entry, items):
+                if entry is not None:
+                    pages.append(entry)
         return pages
 
-    def notion_page_search(self, access_token: str):
-        results = []
-        next_cursor = None
-        has_more = True
+    def _search_all(self) -> list[dict[str, Any]]:
+        """Fetch every page and database shared with the integration in one pass.
 
+        Notion's /v1/search returns both pages and databases when no object filter
+        is supplied, so combining the previous two filtered loops halves the number
+        of search round-trips required to enumerate the workspace.
+        """
+        results: list[dict[str, Any]] = []
+        next_cursor: str | None = None
+        has_more = True
         while has_more:
-            data: dict[str, Any] = {
-                "filter": {"value": "page", "property": "object"},
-                **({"start_cursor": next_cursor} if next_cursor else {}),
-            }
-            headers = {
-                "Content-Type": "application/json",
-                "Authorization": f"Bearer {access_token}",
-                "Notion-Version": self._API_VERSION,
-            }
-
-            response = requests.post(
-                url=self._NOTION_PAGE_SEARCH, json=data, headers=headers, timeout=__TIMEOUT_SECONDS__
-            )
-            response_json = response.json()
-
-            results.extend(response_json.get("results", []))
-
-            has_more = response_json.get("has_more", False)
-            next_cursor = response_json.get("next_cursor", None)
-
+            payload: dict[str, Any] = {}
+            if next_cursor:
+                payload["start_cursor"] = next_cursor
+            data = self._make_request("post", "/search", json_data=payload) or {}
+            results.extend(data.get("results", []))
+            has_more = data.get("has_more", False)
+            next_cursor = data.get("next_cursor")
         return results
 
-    def notion_database_search(self, access_token: str):
-        results = []
-        next_cursor = None
-        has_more = True
+    def _build_page_entry(self, item: dict[str, Any]) -> OnlineDocumentPage | None:
+        obj = item.get("object")
+        if obj == "page":
+            page_name = "Untitled"
+            for key in item.get("properties", {}):
+                prop = item["properties"][key]
+                if "title" in prop and prop["title"]:
+                    title_list = prop["title"]
+                    if len(title_list) > 0 and "plain_text" in title_list[0]:
+                        page_name = title_list[0]["plain_text"]
+            page_type = "page"
+        elif obj == "database":
+            title = item.get("title", [])
+            page_name = title[0]["plain_text"] if len(title) > 0 else "Untitled"
+            page_type = "database"
+        else:
+            return None
+
+        parent = item.get("parent", {})
+        parent_type = parent.get("type")
+        try:
+            if parent_type == "block_id":
+                parent_id = self._resolve_block_parent_page_id(parent[parent_type])
+            elif parent_type == "workspace":
+                parent_id = "root"
+            elif parent_type and parent_type in parent:
+                parent_id = parent[parent_type]
+            else:
+                parent_id = "root"
+        except ValueError:
+            return None
+
+        return OnlineDocumentPage(
+            page_id=item["id"],
+            page_name=page_name,
+            page_icon=None,
+            parent_id=parent_id,
+            type=page_type,
+            last_edited_time=item["last_edited_time"],
+        )
 
-        while has_more:
-            data: dict[str, Any] = {
-                "filter": {"value": "database", "property": "object"},
-                **({"start_cursor": next_cursor} if next_cursor else {}),
-            }
-            headers = {
-                "Content-Type": "application/json",
-                "Authorization": f"Bearer {access_token}",
-                "Notion-Version": self._API_VERSION,
-            }
-            response = requests.post(
-                url=self._NOTION_PAGE_SEARCH, json=data, headers=headers, timeout=__TIMEOUT_SECONDS__
-            )
-            response_json = response.json()
-
-            results.extend(response_json.get("results", []))
-
-            has_more = response_json.get("has_more", False)
-            next_cursor = response_json.get("next_cursor", None)
+    @staticmethod
+    def _resolve_worker_count() -> int:
+        raw = os.environ.get("NOTION_PARENT_RESOLVE_WORKERS")
+        if not raw:
+            return _DEFAULT_PARENT_RESOLVE_WORKERS
+        try:
+            value = int(raw)
+        except ValueError:
+            return _DEFAULT_PARENT_RESOLVE_WORKERS
+        return max(1, min(value, _MAX_PARENT_RESOLVE_WORKERS))
 
-        return results
+    def notion_page_search(self, access_token: str):
+        return [item for item in self._search_all() if item.get("object") == "page"]
+
+    def notion_database_search(self, access_token: str):
+        return [item for item in self._search_all() if item.get("object") == "database"]
 
     def notion_block_parent_page_id(self, access_token: str, block_id: str):
-        headers = {
-            "Authorization": f"Bearer {access_token}",
-            "Notion-Version": self._API_VERSION,
-        }
-        response = requests.get(
-            url=f"{self._NOTION_BLOCK_SEARCH}/{block_id}", headers=headers, timeout=__TIMEOUT_SECONDS__
-        )
-        if response.status_code == 404:
-            # Treat inaccessible parent as a root-level item.
-            return "root"
-        if response.status_code != 200:
-            try:
-                response_json = response.json()
-                message = response_json.get("message", "unknown error")
-            except requests.exceptions.JSONDecodeError:
-                message = response.text or "unknown error"
-            raise ValueError(f"Error fetching block parent page ID: {message}")
-        response_json = response.json()
-        parent = response_json["parent"]
-        parent_type = parent["type"]
-        if parent_type == "block_id":
-            return self.notion_block_parent_page_id(access_token, parent[parent_type])
-        return parent[parent_type]
+        return self._resolve_block_parent_page_id(block_id)
+
+    def _resolve_block_parent_page_id(self, block_id: str) -> str:
+        clean_id = block_id.replace("-", "")
+        cached = self._parent_cache.get(clean_id)
+        if cached is not None:
+            return cached
+
+        data = self._make_request("get", f"/blocks/{clean_id}", allow_status=(404,))
+        if data is None:
+            result = "root"
+        else:
+            parent = data.get("parent", {})
+            parent_type = parent.get("type")
+            if parent_type == "block_id":
+                result = self._resolve_block_parent_page_id(parent[parent_type])
+            elif parent_type == "workspace":
+                result = "root"
+            elif parent_type and parent_type in parent:
+                result = parent[parent_type]
+            else:
+                result = "root"
+
+        with self._parent_cache_lock:
+            self._parent_cache[clean_id] = result
+        return result