Connector Setup

class SourceNotion(YamlDeclarativeSource):
    """
    Main Airbyte source connector for Notion API integration.
    Extends YamlDeclarativeSource to combine declarative YAML configuration
    with custom Python streams for complex operations.
    """
    
    def __init__(self):
        """
        Initializes the connector with manifest.yaml configuration.
        Sets up declarative streams and prepares for custom stream injection.
        """
    
    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
        """
        Returns complete list of available streams including both declarative
        and custom Python streams.
        
        Args:
            config: Connector configuration containing authentication and sync settings
            
        Returns:
            List of Stream objects for data extraction
        """
    
    def _get_authenticator(self, config: Mapping[str, Any]) -> TokenAuthenticator:
        """
        Creates and returns the appropriate authenticator based on configuration.
        Supports OAuth2.0, token-based, and legacy authentication methods.
        
        Args:
            config: Configuration mapping containing credentials
            
        Returns:
            TokenAuthenticator configured for the specified auth method
        """

def run():
    """
    Main entry point for the Airbyte source connector.
    Initializes SourceNotion and launches with command line arguments.
    Used by the source-notion console script.
    """

from source_notion import SourceNotion

# Initialize the connector
source = SourceNotion()

# OAuth2.0 configuration
oauth_config = {
    "credentials": {
        "auth_type": "OAuth2.0",
        "client_id": "your_client_id",
        "client_secret": "your_client_secret",
        "access_token": "your_oauth_token"
    },
    "start_date": "2023-01-01T00:00:00.000Z"
}

# Get available streams
streams = source.streams(oauth_config)
print(f"Available streams: {[stream.name for stream in streams]}")

# Token authentication configuration
token_config = {
    "credentials": {
        "auth_type": "token",
        "token": "secret_integration_token"
    },
    "start_date": "2023-01-01T00:00:00.000Z"
}

# Initialize and configure
source = SourceNotion()
streams = source.streams(token_config)

# Test authenticator creation
authenticator = source._get_authenticator(token_config)

# Legacy format (backward compatibility)
legacy_config = {
    "access_token": "notion_token",
    "start_date": "2023-01-01T00:00:00.000Z"
}

source = SourceNotion()
authenticator = source._get_authenticator(legacy_config)

# Using as installed console script
source-notion spec
source-notion check --config config.json
source-notion discover --config config.json
source-notion read --config config.json --catalog catalog.json

# Using Python module directly
python -m source_notion.run spec
python -m source_notion.run check --config config.json

from airbyte_cdk.models import AirbyteConnectionStatus, Status
from airbyte_cdk.logger import AirbyteLogger

logger = AirbyteLogger()
source = SourceNotion()

# Test connection
config = {
    "credentials": {
        "auth_type": "token",
        "token": "your_token"
    }
}

try:
    status = source.check(logger, config)
    if status.status == Status.SUCCEEDED:
        print("Connection successful!")
    else:
        print(f"Connection failed: {status.message}")
except Exception as e:
    print(f"Error testing connection: {e}")