Error Handling

class RequestTimeoutError(Exception):
    """Exception for requests that timeout."""
    
    code: str = "notionhq_client_request_timeout"
    
    def __init__(self, message="Request to Notion API has timed out"):
        """
        Initialize timeout error.
        
        Parameters:
        - message: str, error message (default provided)
        """

class HTTPResponseError(Exception):
    """Exception for HTTP errors."""
    
    code: str = "notionhq_client_response_error"
    status: int
    headers: httpx.Headers
    body: str
    
    def __init__(self, response, message=None):
        """
        Initialize HTTP response error.
        
        Parameters:
        - response: httpx.Response, the failed HTTP response
        - message: str, optional custom error message
        """

class APIResponseError(HTTPResponseError):
    """An error raised by Notion API."""
    
    code: APIErrorCode
    
    def __init__(self, response, message, code):
        """
        Initialize API response error.
        
        Parameters:
        - response: httpx.Response, the failed HTTP response
        - message: str, error message from API
        - code: APIErrorCode, specific error code from API
        """

class APIErrorCode(str, Enum):
    """API error code enumeration."""
    
    Unauthorized = "unauthorized"
    """The bearer token is not valid."""
    
    RestrictedResource = "restricted_resource"
    """Given the bearer token used, the client doesn't have permission to perform this operation."""
    
    ObjectNotFound = "object_not_found"
    """Given the bearer token used, the resource does not exist. This error can also indicate that the resource has not been shared with owner of the bearer token."""
    
    RateLimited = "rate_limited"
    """This request exceeds the number of requests allowed. Slow down and try again."""
    
    InvalidJSON = "invalid_json"
    """The request body could not be decoded as JSON."""
    
    InvalidRequestURL = "invalid_request_url"
    """The request URL is not valid."""
    
    InvalidRequest = "invalid_request"
    """This request is not supported."""
    
    ValidationError = "validation_error"
    """The request body does not match the schema for the expected parameters."""
    
    ConflictError = "conflict_error"
    """The transaction could not be completed, potentially due to a data collision. Make sure the parameters are up to date and try again."""
    
    InternalServerError = "internal_server_error"
    """An unexpected error occurred. Reach out to Notion support."""
    
    ServiceUnavailable = "service_unavailable"
    """Notion is unavailable. Try again later. This can occur when the time to respond to a request takes longer than 60 seconds, the maximum request timeout."""

def is_api_error_code(code):
    """
    Check if given code belongs to the list of valid API error codes.
    
    Parameters:
    - code: str, error code to check
    
    Returns:
    bool, True if code is a valid API error code
    """

from notion_client import Client, APIResponseError, APIErrorCode

notion = Client(auth="your_token")

try:
    page = notion.pages.retrieve(page_id="invalid_page_id")
except APIResponseError as error:
    print(f"API Error: {error}")
    print(f"Error Code: {error.code}")
    print(f"HTTP Status: {error.status}")
    print(f"Response Body: {error.body}")

from notion_client import Client, APIResponseError, APIErrorCode
import logging

try:
    response = notion.databases.query(
        database_id="database_id_here",
        filter={"property": "Status", "select": {"equals": "Active"}}
    )
except APIResponseError as error:
    if error.code == APIErrorCode.ObjectNotFound:
        print("Database not found or not accessible")
    elif error.code == APIErrorCode.Unauthorized:
        print("Invalid or expired authentication token")
    elif error.code == APIErrorCode.RestrictedResource:
        print("Insufficient permissions to access this resource")
    elif error.code == APIErrorCode.RateLimited:
        print("Rate limit exceeded, please slow down requests")
    elif error.code == APIErrorCode.ValidationError:
        print("Invalid request parameters")
        print(f"Details: {error.body}")
    else:
        logging.error(f"Unexpected API error: {error.code} - {error}")

from notion_client import (
    Client, 
    APIResponseError, 
    HTTPResponseError, 
    RequestTimeoutError,
    APIErrorCode
)
import time
import logging

def robust_api_call(notion, operation, max_retries=3):
    """
    Make an API call with comprehensive error handling and retries.
    
    Parameters:
    - notion: Client instance
    - operation: callable that performs the API operation
    - max_retries: int, maximum number of retry attempts
    
    Returns:
    API response or None if all retries failed
    """
    for attempt in range(max_retries + 1):
        try:
            return operation()
            
        except APIResponseError as error:
            if error.code == APIErrorCode.RateLimited:
                if attempt < max_retries:
                    wait_time = 2 ** attempt  # Exponential backoff
                    print(f"Rate limited. Waiting {wait_time} seconds before retry...")
                    time.sleep(wait_time)
                    continue
                else:
                    print("Max retries reached for rate limiting")
                    raise
                    
            elif error.code == APIErrorCode.InternalServerError:
                if attempt < max_retries:
                    wait_time = 2 ** attempt
                    print(f"Server error. Retrying in {wait_time} seconds...")
                    time.sleep(wait_time)
                    continue
                else:
                    print("Max retries reached for server errors")
                    raise
                    
            elif error.code == APIErrorCode.ServiceUnavailable:
                if attempt < max_retries:
                    wait_time = 5 * (attempt + 1)  # Linear backoff for service issues
                    print(f"Service unavailable. Retrying in {wait_time} seconds...")
                    time.sleep(wait_time)
                    continue
                else:
                    print("Max retries reached for service unavailability")
                    raise
                    
            elif error.code in [
                APIErrorCode.Unauthorized,
                APIErrorCode.ObjectNotFound,
                APIErrorCode.RestrictedResource,
                APIErrorCode.ValidationError
            ]:
                # These errors won't be resolved by retrying
                print(f"Non-retryable error: {error.code}")
                raise
                
            else:
                print(f"Unknown API error: {error.code}")
                raise
                
        except RequestTimeoutError:
            if attempt < max_retries:
                wait_time = 2 ** attempt
                print(f"Request timeout. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                print("Max retries reached for timeouts")
                raise
                
        except HTTPResponseError as error:
            print(f"HTTP error {error.status}: {error}")
            if error.status >= 500 and attempt < max_retries:
                # Retry server errors
                wait_time = 2 ** attempt
                print(f"Server error. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                raise
                
        except Exception as error:
            print(f"Unexpected error: {error}")
            raise
    
    return None

# Usage example
notion = Client(auth="your_token")

def get_database():
    return notion.databases.retrieve(database_id="database_id_here")

database = robust_api_call(notion, get_database)
if database:
    print("Successfully retrieved database")
else:
    print("Failed to retrieve database after all retries")

import logging
from notion_client import Client, APIResponseError, HTTPResponseError, RequestTimeoutError

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

notion = Client(
    auth="your_token",
    log_level=logging.DEBUG  # Enable debug logging for API requests
)

def log_api_error(error, operation_name):
    """Log API errors with relevant details."""
    if isinstance(error, APIResponseError):
        logger.error(
            f"API Error in {operation_name}: "
            f"Code={error.code}, Status={error.status}, "
            f"Message={str(error)}"
        )
        if error.code == APIErrorCode.ValidationError:
            logger.error(f"Validation details: {error.body}")
    elif isinstance(error, RequestTimeoutError):
        logger.error(f"Timeout in {operation_name}: {str(error)}")
    elif isinstance(error, HTTPResponseError):
        logger.error(
            f"HTTP Error in {operation_name}: "
            f"Status={error.status}, Message={str(error)}"
        )
    else:
        logger.error(f"Unexpected error in {operation_name}: {str(error)}")

try:
    users = notion.users.list()
    logger.info(f"Successfully retrieved {len(users['results'])} users")
except (APIResponseError, HTTPResponseError, RequestTimeoutError) as error:
    log_api_error(error, "users.list")

from notion_client import Client, APIResponseError, APIErrorCode

class NotionClientWrapper:
    """Wrapper class with enhanced error handling and context."""
    
    def __init__(self, auth_token):
        self.notion = Client(auth=auth_token)
        self.last_error = None
        
    def safe_database_query(self, database_id, **kwargs):
        """Query database with error context and fallback strategies."""
        try:
            return self.notion.databases.query(database_id=database_id, **kwargs)
            
        except APIResponseError as error:
            self.last_error = error
            
            if error.code == APIErrorCode.ObjectNotFound:
                print(f"Database {database_id} not found. Checking if it exists...")
                return self._handle_not_found_database(database_id)
                
            elif error.code == APIErrorCode.ValidationError:
                print("Query validation failed. Trying with simplified query...")
                return self._retry_with_simple_query(database_id)
                
            elif error.code == APIErrorCode.RestrictedResource:
                print("Access restricted. Trying to get basic database info...")
                return self._get_basic_database_info(database_id)
                
            else:
                raise  # Re-raise unhandled errors
    
    def _handle_not_found_database(self, database_id):
        """Handle database not found scenario."""
        try:
            # Try to retrieve the database directly
            db_info = self.notion.databases.retrieve(database_id=database_id)
            print("Database exists but may be empty")
            return {"results": [], "database_info": db_info}
        except APIResponseError:
            print("Database truly does not exist or is not accessible")
            return None
    
    def _retry_with_simple_query(self, database_id):
        """Retry with a simpler query."""
        try:
            return self.notion.databases.query(
                database_id=database_id,
                page_size=10  # Reduce page size and remove filters
            )
        except APIResponseError:
            print("Even simple query failed")
            return None
    
    def _get_basic_database_info(self, database_id):
        """Get basic database information when full access is restricted."""
        try:
            db_info = self.notion.databases.retrieve(database_id=database_id)
            return {"results": [], "database_info": db_info, "access_limited": True}
        except APIResponseError:
            return None

# Usage
wrapper = NotionClientWrapper("your_token")
result = wrapper.safe_database_query("database_id_here")

if result is None:
    print("Could not access database")
elif "access_limited" in result:
    print("Limited access - only database metadata available")
else:
    print(f"Retrieved {len(result['results'])} pages")

import re
from notion_client import Client

def is_valid_notion_id(notion_id):
    """Check if a string is a valid Notion ID format."""
    # Notion IDs are UUIDs with or without hyphens
    uuid_pattern = r'^[0-9a-f]{8}-?[0-9a-f]{4}-?[0-9a-f]{4}-?[0-9a-f]{4}-?[0-9a-f]{12}$'
    return re.match(uuid_pattern, notion_id.lower()) is not None

def safe_page_retrieve(notion, page_id):
    """Safely retrieve a page with pre-validation."""
    if not is_valid_notion_id(page_id):
        raise ValueError(f"Invalid page ID format: {page_id}")
    
    try:
        return notion.pages.retrieve(page_id=page_id)
    except APIResponseError as error:
        if error.code == APIErrorCode.ObjectNotFound:
            print(f"Page {page_id} does not exist or is not accessible")
        raise

import time
from collections import deque
from notion_client import Client

class RateLimitedClient:
    """Client wrapper with built-in rate limiting."""
    
    def __init__(self, auth_token, requests_per_second=3):
        self.notion = Client(auth=auth_token)
        self.requests_per_second = requests_per_second
        self.request_times = deque()
    
    def _enforce_rate_limit(self):
        """Enforce rate limiting before making requests."""
        now = time.time()
        
        # Remove requests older than 1 second
        while self.request_times and now - self.request_times[0] > 1.0:
            self.request_times.popleft()
        
        # If we've made too many requests, wait
        if len(self.request_times) >= self.requests_per_second:
            sleep_time = 1.0 - (now - self.request_times[0])
            if sleep_time > 0:
                time.sleep(sleep_time)
        
        self.request_times.append(now)
    
    def query_database(self, database_id, **kwargs):
        """Query database with rate limiting."""
        self._enforce_rate_limit()
        return self.notion.databases.query(database_id=database_id, **kwargs)

# Usage
rate_limited_notion = RateLimitedClient("your_token", requests_per_second=2)
results = rate_limited_notion.query_database("database_id_here")