Core Connector API

from typing import Tuple
from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource

class SourceXero(YamlDeclarativeSource):
    """
    Main Airbyte source connector for Xero API.
    
    Inherits from YamlDeclarativeSource to provide declarative configuration
    support using manifest.yaml for stream definitions.
    """
    
    def __init__(self):
        """
        Initialize the Xero source connector.
        
        Automatically loads configuration from manifest.yaml file in the
        source_xero package directory.
        """
    
    def check_connection(self, logger, config: dict) -> Tuple[bool, str]:
        """
        Test connection to Xero API using provided configuration.
        
        Args:
            logger: Python logger instance for diagnostic output
            config (dict): Connection configuration containing access_token,
                         tenant_id, and start_date
        
        Returns:
            Tuple[bool, str]: (success_status, error_message)
                            - (True, None) for successful connection
                            - (False, error_details) for failed connection
        
        Raises:
            Various HTTP and authentication exceptions based on Xero API response
        """
    
    def streams(self, config: dict) -> list:
        """
        Discover and return all available data streams for the connector.
        
        Args:
            config (dict): Connection configuration
            
        Returns:
            list: List of configured stream objects (21 total streams)
                 Each stream object contains metadata, schema, and sync methods
        """

def run():
    """
    Main entry point for running the Xero source connector.
    
    Creates a SourceXero instance and launches it using Airbyte CDK's
    launch function with command-line arguments from sys.argv.
    
    This function is registered as the 'source-xero' console script
    in the package's pyproject.toml configuration.
    
    Usage:
        Called automatically when running 'source-xero' command or
        when executing the main.py module directly.
    """

from source_xero import SourceXero
import logging

# Create logger
logger = logging.getLogger(__name__)

# Initialize connector
source = SourceXero()

# Configuration with required fields
config = {
    "access_token": "your_xero_access_token_here",
    "tenant_id": "your_xero_tenant_id_here", 
    "start_date": "2023-01-01T00:00:00Z"
}

# Test connection
is_connected, error = source.check_connection(logger, config)
if is_connected:
    print("Successfully connected to Xero API")
else:
    print(f"Connection failed: {error}")

from source_xero import SourceXero

source = SourceXero()
config = {
    "access_token": "your_token",
    "tenant_id": "your_tenant", 
    "start_date": "2023-01-01T00:00:00Z"
}

# Get all available streams
streams = source.streams(config)
print(f"Total available streams: {len(streams)}")

# Stream information
for stream in streams:
    print(f"Stream Name: {stream.name}")
    print(f"Primary Key: {getattr(stream, 'primary_key', 'None')}")
    print(f"Supports Incremental: {hasattr(stream, 'incremental_sync')}")
    print("---")

# Expected output includes 21 streams:
# bank_transactions, contacts, credit_notes, invoices, manual_journals,
# overpayments, prepayments, purchase_orders, accounts, bank_transfers,
# employees, items, payments, users, branding_themes, contact_groups,
# currencies, organisations, repeating_invoices, tax_rates, tracking_categories

from source_xero.run import run
import sys

# Simulate command line arguments
sys.argv = ['source-xero', 'check', '--config', 'config.json']

# Run the connector
run()  # This will execute the check command