or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

configuration.mdcore-connector.mddata-processing.mddata-streams.mdindex.md
tile.json

tessl/pypi-airbyte-source-xero

Airbyte source connector for extracting data from the Xero accounting API with support for 21 data streams and incremental sync capabilities

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/airbyte-source-xero@1.0.x

To install, run

npx @tessl/cli install tessl/pypi-airbyte-source-xero@1.0.0

index.mddocs/

Airbyte Source Xero

An Airbyte source connector for extracting data from the Xero accounting API. This connector enables automated data synchronization from Xero's comprehensive accounting platform, supporting 21 different data streams including transactions, contacts, invoices, and financial records with incremental sync capabilities.

Package Information

  • Package Name: airbyte-source-xero
  • Package Type: pypi
  • Language: Python
  • Installation: pip install airbyte-source-xero
  • Airbyte Framework: Declarative low-code CDK v1
  • Python Version: 3.9 - 3.11

Core Imports

from source_xero import SourceXero
from source_xero.run import run

For date processing utilities:

from source_xero.components import ParseDates, CustomExtractor

Basic Usage

Command Line Usage

# Run as Airbyte connector
source-xero discover --config config.json
source-xero check --config config.json
source-xero read --config config.json --catalog catalog.json

Programmatic Usage

from source_xero import SourceXero
import json

# Load configuration
config = {
    "access_token": "your_xero_access_token",
    "tenant_id": "your_xero_tenant_id", 
    "start_date": "2023-01-01T00:00:00Z"
}

# Initialize source connector
source = SourceXero()

# Check connection
status, error = source.check_connection(logger, config)
if status:
    print("Connection successful")
    
    # Get available streams
    streams = source.streams(config)
    print(f"Available streams: {len(streams)}")
    
    # Stream names: bank_transactions, contacts, invoices, etc.
    for stream in streams:
        print(f"Stream: {stream.name}")

Date Processing

from source_xero.components import ParseDates

# Parse Xero .NET JSON date format
xero_date = "/Date(1419937200000+0000)/"
parsed_date = ParseDates.parse_date(xero_date)
print(parsed_date)  # 2014-12-30 07:00:00+00:00

# Convert dates in nested data structures
data = {"UpdatedDate": "/Date(1419937200000+0000)/", "Amount": 100.50}
ParseDates.convert_dates(data)
print(data)  # {"UpdatedDate": "2014-12-30T07:00:00+00:00", "Amount": 100.50}

Architecture

The connector is built using Airbyte's declarative low-code framework with the following key components:

  • SourceXero: Main connector class inheriting from YamlDeclarativeSource
  • Manifest Configuration: YAML-based stream definitions with authentication, pagination, and incremental sync
  • Custom Components: Date parsing and record extraction for Xero-specific data formats
  • Stream Processors: 21 pre-configured data streams with automatic schema validation

Capabilities

Core Connector API

Main connector functionality including initialization, connection testing, and stream management. Provides the primary interface for Airbyte framework integration.

class SourceXero(YamlDeclarativeSource):
    def __init__(self): ...
    def check_connection(self, logger, config: dict) -> tuple[bool, str]: ...
    def streams(self, config: dict) -> list: ...

def run(): ...

Core Connector API

Data Processing Components

Specialized utilities for handling Xero-specific data formats, particularly .NET JSON date strings and custom record extraction with automatic date conversion.

class ParseDates:
    @staticmethod
    def parse_date(value: str) -> datetime | None: ...
    @staticmethod
    def convert_dates(obj: dict | list) -> None: ...

@dataclass
class CustomExtractor(RecordExtractor):
    field_path: list[str | InterpolatedString]
    config: Config
    decoder: Decoder
    def extract_records(self, response: requests.Response) -> list[dict]: ...

Data Processing

Configuration & Authentication

Configuration specification and authentication setup for connecting to Xero API with proper credentials and tenant management.

# Configuration Schema
{
    "access_token": str,  # Required, secret
    "tenant_id": str,     # Required, secret  
    "start_date": str     # Required, ISO 8601 format
}

Configuration & Authentication

Data Streams

Access to 21 different Xero data streams organized into transactional data (with incremental sync) and reference data (snapshot sync).

Transactional Streams (9 streams with incremental sync):

  • bank_transactions, contacts, credit_notes, invoices, manual_journals, overpayments, prepayments, purchase_orders, payments

Reference Streams (12 streams with snapshot sync):

  • accounts, bank_transfers, employees, items, users, branding_themes, contact_groups, currencies, organisations, repeating_invoices, tax_rates, tracking_categories

Data Streams

Configuration

# Required configuration fields
ConfigSpec = {
    "access_token": {
        "type": "string",
        "title": "Access Token", 
        "description": "Enter your Xero application's access token",
        "airbyte_secret": True
    },
    "tenant_id": {
        "type": "string",
        "title": "Tenant ID",
        "description": "Enter your Xero organization's Tenant ID", 
        "airbyte_secret": True
    },
    "start_date": {
        "type": "string",
        "title": "Start Date",
        "description": "UTC date and time in format YYYY-MM-DDTHH:mm:ssZ",
        "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$",
        "format": "date-time"
    }
}

Error Handling

The connector implements comprehensive error handling for common Xero API scenarios:

  • 401 Unauthorized: Authentication failures with clear error messages
  • 403 Forbidden: Permission issues (records ignored)
  • 429 Rate Limited: Automatic retry with 30-second backoff
  • Date Parsing Errors: Graceful handling of malformed date strings
  • Network Issues: Standard HTTP client error handling with retries