Spaces:
Running
Running
| from typing import Any | |
| import requests | |
| from langchain.tools import StructuredTool | |
| from pydantic import BaseModel, Field | |
| from langflow.base.langchain_utilities.model import LCToolComponent | |
| from langflow.field_typing import Tool | |
| from langflow.inputs import DropdownInput, SecretStrInput, StrInput | |
| from langflow.schema import Data | |
| class NotionSearch(LCToolComponent): | |
| display_name: str = "Search " | |
| description: str = "Searches all pages and databases that have been shared with an integration." | |
| documentation: str = "https://docs.langflow.org/integrations/notion/search" | |
| icon = "NotionDirectoryLoader" | |
| inputs = [ | |
| SecretStrInput( | |
| name="notion_secret", | |
| display_name="Notion Secret", | |
| info="The Notion integration token.", | |
| required=True, | |
| ), | |
| StrInput( | |
| name="query", | |
| display_name="Search Query", | |
| info="The text that the API compares page and database titles against.", | |
| ), | |
| DropdownInput( | |
| name="filter_value", | |
| display_name="Filter Type", | |
| info="Limits the results to either only pages or only databases.", | |
| options=["page", "database"], | |
| value="page", | |
| ), | |
| DropdownInput( | |
| name="sort_direction", | |
| display_name="Sort Direction", | |
| info="The direction to sort the results.", | |
| options=["ascending", "descending"], | |
| value="descending", | |
| ), | |
| ] | |
| class NotionSearchSchema(BaseModel): | |
| query: str = Field(..., description="The search query text.") | |
| filter_value: str = Field(default="page", description="Filter type: 'page' or 'database'.") | |
| sort_direction: str = Field(default="descending", description="Sort direction: 'ascending' or 'descending'.") | |
| def run_model(self) -> list[Data]: | |
| results = self._search_notion(self.query, self.filter_value, self.sort_direction) | |
| records = [] | |
| combined_text = f"Results found: {len(results)}\n\n" | |
| for result in results: | |
| result_data = { | |
| "id": result["id"], | |
| "type": result["object"], | |
| "last_edited_time": result["last_edited_time"], | |
| } | |
| if result["object"] == "page": | |
| result_data["title_or_url"] = result["url"] | |
| text = f"id: {result['id']}\ntitle_or_url: {result['url']}\n" | |
| elif result["object"] == "database": | |
| if "title" in result and isinstance(result["title"], list) and len(result["title"]) > 0: | |
| result_data["title_or_url"] = result["title"][0]["plain_text"] | |
| text = f"id: {result['id']}\ntitle_or_url: {result['title'][0]['plain_text']}\n" | |
| else: | |
| result_data["title_or_url"] = "N/A" | |
| text = f"id: {result['id']}\ntitle_or_url: N/A\n" | |
| text += f"type: {result['object']}\nlast_edited_time: {result['last_edited_time']}\n\n" | |
| combined_text += text | |
| records.append(Data(text=text, data=result_data)) | |
| self.status = records | |
| return records | |
| def build_tool(self) -> Tool: | |
| return StructuredTool.from_function( | |
| name="notion_search", | |
| description="Search Notion pages and databases. " | |
| "Input should include the search query and optionally filter type and sort direction.", | |
| func=self._search_notion, | |
| args_schema=self.NotionSearchSchema, | |
| ) | |
| def _search_notion( | |
| self, query: str, filter_value: str = "page", sort_direction: str = "descending" | |
| ) -> list[dict[str, Any]]: | |
| url = "https://api.notion.com/v1/search" | |
| headers = { | |
| "Authorization": f"Bearer {self.notion_secret}", | |
| "Content-Type": "application/json", | |
| "Notion-Version": "2022-06-28", | |
| } | |
| data = { | |
| "query": query, | |
| "filter": {"value": filter_value, "property": "object"}, | |
| "sort": {"direction": sort_direction, "timestamp": "last_edited_time"}, | |
| } | |
| response = requests.post(url, headers=headers, json=data, timeout=10) | |
| response.raise_for_status() | |
| results = response.json() | |
| return results["results"] | |