import os import logging from typing import Dict, Any, List, Optional from dotenv import load_dotenv from notion_client import Client from notion_client.errors import APIResponseError # Load environment variables load_dotenv() # Set up logging with colors logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class NotionConnectionTester: """Test Notion API connection and permissions.""" def __init__(self, token: str): """Initialize the Notion client with the provided token.""" self.client = Client(auth=token) self.token = token def test_authentication(self) -> bool: """Test if the Notion token is valid and can authenticate.""" try: logger.info("🔐 Testing authentication...") # Try to get the bot user info response = self.client.users.me() if response: logger.info(f"✅ Authentication successful!") logger.info(f" Bot name: {response.get('name', 'Unknown')}") logger.info(f" Bot type: {response.get('type', 'Unknown')}") logger.info(f" Bot ID: {response.get('id', 'Unknown')}") return True else: logger.error("❌ Authentication failed - no response") return False except APIResponseError as e: logger.error(f"❌ Authentication failed - API Error: {e}") return False except Exception as e: logger.error(f"❌ Authentication failed - Unexpected error: {e}") return False def list_accessible_databases(self) -> List[Dict[str, Any]]: """List all databases accessible to the integration.""" try: logger.info("📊 Fetching accessible databases...") response = self.client.search( filter={"property": "object", "value": "database"} ) databases = response.get('results', []) if databases: logger.info(f"✅ Found {len(databases)} accessible database(s):") for i, db in enumerate(databases, 1): title = self._extract_title(db.get('title', [])) logger.info(f" {i}. {title} (ID: {db['id']})") else: logger.warning("⚠️ No accessible databases found") return databases except APIResponseError as e: logger.error(f"❌ Failed to fetch databases - API Error: {e}") return [] except Exception as e: logger.error(f"❌ Failed to fetch databases - Unexpected error: {e}") return [] def list_accessible_pages(self) -> List[Dict[str, Any]]: """List all pages accessible to the integration.""" try: logger.info("📄 Fetching accessible pages...") response = self.client.search( filter={"property": "object", "value": "page"} ) pages = response.get('results', []) if pages: logger.info(f"✅ Found {len(pages)} accessible page(s):") for i, page in enumerate(pages, 1): title = self._extract_title(page.get('properties', {}).get('title', {}).get('title', [])) if not title: title = "Untitled" logger.info(f" {i}. {title} (ID: {page['id']})") else: logger.warning("⚠️ No accessible pages found") return pages except APIResponseError as e: logger.error(f"❌ Failed to fetch pages - API Error: {e}") return [] except Exception as e: logger.error(f"❌ Failed to fetch pages - Unexpected error: {e}") return [] def test_specific_page_access(self, page_id: str) -> bool: """Test access to a specific page by ID.""" try: logger.info(f"🔍 Testing access to specific page: {page_id}") # Clean page_id (remove dashes if present) clean_page_id = page_id.replace('-', '') # Try to retrieve the page page = self.client.pages.retrieve(clean_page_id) if page: title = self._extract_title(page.get('properties', {}).get('title', {}).get('title', [])) if not title: title = "Untitled" logger.info(f"✅ Successfully accessed page: {title}") logger.info(f" Page ID: {page['id']}") logger.info(f" Created: {page.get('created_time', 'Unknown')}") logger.info(f" Last edited: {page.get('last_edited_time', 'Unknown')}") # Try to get blocks from the page try: blocks_response = self.client.blocks.children.list(block_id=clean_page_id) blocks = blocks_response.get('results', []) logger.info(f" Blocks accessible: {len(blocks)}") except Exception as block_error: logger.warning(f" ⚠️ Could not access blocks: {block_error}") return True else: logger.error("❌ Page access failed - no response") return False except APIResponseError as e: logger.error(f"❌ Page access failed - API Error: {e}") if "Could not find page" in str(e): logger.error(" This usually means the page doesn't exist or the integration doesn't have access to it") return False except Exception as e: logger.error(f"❌ Page access failed - Unexpected error: {e}") return False def test_search_functionality(self) -> bool: """Test general search functionality.""" try: logger.info("🔎 Testing search functionality...") response = self.client.search() results = response.get('results', []) logger.info(f"✅ Search successful - found {len(results)} total items") # Categorize results pages = [r for r in results if r.get('object') == 'page'] databases = [r for r in results if r.get('object') == 'database'] logger.info(f" Pages: {len(pages)}") logger.info(f" Databases: {len(databases)}") return True except APIResponseError as e: logger.error(f"❌ Search failed - API Error: {e}") return False except Exception as e: logger.error(f"❌ Search failed - Unexpected error: {e}") return False def _extract_title(self, title_array: List[Dict[str, Any]]) -> str: """Extract plain text from Notion title array.""" if not title_array: return "" return ''.join([item.get('plain_text', '') for item in title_array]) def run_all_tests(self, specific_page_id: Optional[str] = None) -> Dict[str, bool]: """Run all connection tests and return results.""" logger.info("🚀 Starting Notion API connection tests...") logger.info("=" * 60) results = {} # Test 1: Authentication results['authentication'] = self.test_authentication() logger.info("") if not results['authentication']: logger.error("❌ Authentication failed - stopping tests") return results # Test 2: Search functionality results['search'] = self.test_search_functionality() logger.info("") # Test 3: List databases databases = self.list_accessible_databases() results['databases'] = len(databases) > 0 logger.info("") # Test 4: List pages pages = self.list_accessible_pages() results['pages'] = len(pages) > 0 logger.info("") # Test 5: Specific page access (if provided) if specific_page_id: results['specific_page'] = self.test_specific_page_access(specific_page_id) logger.info("") # Summary logger.info("📋 Test Summary:") logger.info("=" * 60) for test_name, passed in results.items(): status = "✅ PASS" if passed else "❌ FAIL" logger.info(f" {test_name.replace('_', ' ').title()}: {status}") return results def main(): """Main function to run the Notion connection tests.""" # Check if NOTION_TOKEN is set notion_token = os.getenv('NOTION_TOKEN') if not notion_token: logger.error("❌ NOTION_TOKEN environment variable is not set!") logger.error("Please create a .env file with your Notion integration token:") logger.error("NOTION_TOKEN=your_notion_integration_token_here") return # Initialize tester tester = NotionConnectionTester(notion_token) # Ask for specific page ID to test (optional) print("\n" + "=" * 60) print("NOTION API CONNECTION TESTER") print("=" * 60) specific_page_id = input("\nEnter a specific page ID to test (or press Enter to skip): ").strip() if not specific_page_id: specific_page_id = None print("\n") # Run all tests results = tester.run_all_tests(specific_page_id) # Final recommendations print("\n" + "=" * 60) print("RECOMMENDATIONS:") print("=" * 60) if not results.get('authentication'): print("❌ Check your NOTION_TOKEN - it appears to be invalid") elif not results.get('pages') and not results.get('databases'): print("⚠️ Your integration has no access to pages or databases") print(" Make sure you've shared the relevant pages/databases with your integration") elif specific_page_id and not results.get('specific_page'): print(f"❌ The specific page {specific_page_id} is not accessible") print(" Make sure this page is shared with your integration") else: print("✅ Connection looks good! Check the logs above for details") if __name__ == '__main__': main()