0
# Source Configuration
1

2
Core source configuration and connection validation functionality for the Airbyte Source Xero connector. This module handles OAuth2 authentication setup, connection testing, and stream discovery.
3

4
## Core Imports
5

6
```python
7
from typing import Any, List, Mapping, Tuple
8
from airbyte_cdk.sources import AbstractSource
9
from airbyte_cdk.sources.streams import Stream
10
from source_xero.oauth import XeroSingleUseRefreshTokenOauth2Authenticator
11
```
12

13
## Capabilities
14

15
### Source Class
16

17
The main `SourceXero` class that implements the Airbyte source interface for connecting to and extracting data from Xero.
18

19
```python { .api }
20
class SourceXero(AbstractSource):
21
    """
22
    Airbyte source connector for Xero accounting data.
23
    
24
    Inherits from airbyte_cdk.sources.AbstractSource and implements
25
    the required methods for connection testing and stream discovery.
26
    """
27
    
28
    def check_connection(self, logger, config) -> Tuple[bool, any]:
29
        """
30
        Test connection to Xero API by attempting to fetch Organizations.
31
        
32
        Parameters:
33
        - logger: Logger instance for connection status reporting
34
        - config: Configuration dict with authentication and connection details
35
        
36
        Returns:
37
        Tuple of (success: bool, error_message: any)
38
        - success: True if connection succeeds, False otherwise
39
        - error_message: None on success, error details on failure
40
        """
41
    
42
    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
43
        """
44
        Discover and return all available Xero data streams.
45
        
46
        Parameters:
47
        - config: Configuration mapping with authentication details
48
        
49
        Returns:
50
        List of 21 stream instances (14 incremental, 7 full-refresh):
51
        - Incremental: Accounts, BankTransactions, BankTransfers, Contacts, 
52
          CreditNotes, Employees, Invoices, Items, ManualJournals, 
53
          Overpayments, Payments, Prepayments, PurchaseOrders, Users
54
        - Full-refresh: BrandingThemes, ContactGroups, Currencies, 
55
          Organisations, RepeatingInvoices, TaxRates, TrackingCategories
56
        """
57
    
58
    @staticmethod
59
    def get_authenticator(config: Mapping[str, Any]) -> XeroSingleUseRefreshTokenOauth2Authenticator:
60
        """
61
        Create and configure OAuth2 authenticator for Xero API.
62
        
63
        Parameters:
64
        - config: Configuration mapping containing OAuth2 credentials
65
        
66
        Returns:
67
        XeroSingleUseRefreshTokenOauth2Authenticator instance configured with:
68
        - token_refresh_endpoint: "https://identity.xero.com/connect/token"
69
        - client_id: From config["authentication"]["client_id"]
70
        - client_secret: From config["authentication"]["client_secret"]
71
        - access_token_config_path: ["authentication", "access_token"]
72
        - refresh_token_config_path: ["authentication", "refresh_token"]
73
        - token_expiry_date_config_path: ["authentication", "token_expiry_date"]
74
        """
75
```
76

77
### Configuration Validation
78

79
Internal method for validating and transforming configuration parameters.
80

81
```python { .api }
82
def _validate_and_transform(self, config: Mapping[str, Any]) -> Mapping[str, Any]:
83
    """
84
    Validate start_date format and transform configuration.
85
    
86
    Parameters:
87
    - config: Raw configuration mapping
88
    
89
    Returns:
90
    Validated and transformed configuration mapping
91
    
92
    Raises:
93
    ValueError: If start_date format is invalid
94
    """
95
```
96

97
## Configuration Requirements
98

99
### Authentication Configuration
100

101
```python { .api }
102
AuthenticationConfig = {
103
    "client_id": str,           # Xero application Client ID
104
    "client_secret": str,       # Xero application Client Secret
105
    "refresh_token": str,       # OAuth2 refresh token
106
    "access_token": str,        # OAuth2 access token  
107
    "token_expiry_date": str    # Token expiry in ISO format
108
}
109
```
110

111
### Connection Configuration
112

113
```python { .api }
114
ConnectionConfig = {
115
    "authentication": AuthenticationConfig,  # OAuth2 credentials
116
    "tenant_id": str,                       # Xero organization tenant ID
117
    "start_date": str                       # UTC date in YYYY-MM-DDTHH:mm:ssZ
118
}
119
```
120

121
## Usage Examples
122

123
### Basic Connection Test
124

125
```python
126
from source_xero import SourceXero
127
import logging
128

129
logger = logging.getLogger("airbyte")
130
config = {
131
    "authentication": {
132
        "client_id": "your-client-id",
133
        "client_secret": "your-client-secret", 
134
        "refresh_token": "your-refresh-token",
135
        "access_token": "your-access-token",
136
        "token_expiry_date": "2024-12-31T23:59:59Z"
137
    },
138
    "tenant_id": "your-tenant-id",
139
    "start_date": "2023-01-01T00:00:00Z"
140
}
141

142
source = SourceXero()
143
is_connected, error = source.check_connection(logger, config)
144

145
if is_connected:
146
    print("Successfully connected to Xero")
147
    streams = source.streams(config)
148
    print(f"Found {len(streams)} available streams")
149
else:
150
    print(f"Connection failed: {error}")
151
```
152

153
### Stream Discovery
154

155
```python
156
source = SourceXero()
157
streams = source.streams(config)
158

159
# List all stream names
160
stream_names = [stream.name for stream in streams]
161
print("Available streams:", stream_names)
162

163
# Separate incremental and full-refresh streams
164
incremental_streams = [s for s in streams if hasattr(s, 'cursor_field')]
165
full_refresh_streams = [s for s in streams if not hasattr(s, 'cursor_field')]
166

167
print(f"Incremental streams: {len(incremental_streams)}")
168
print(f"Full-refresh streams: {len(full_refresh_streams)}")
169
```
170

171
### Authentication Setup
172

173
```python
174
# Get configured authenticator
175
authenticator = SourceXero.get_authenticator(config)
176

177
# The authenticator handles:
178
# - OAuth2 token refresh
179
# - Rate limiting and retries
180
# - Request header management
181
# - Error handling for 401/429/500+ responses
182
```
183

184
## Error Handling
185

186
### Connection Errors
187

188
The `check_connection` method catches and reports various connection issues:
189

190
- **Invalid Credentials**: OAuth2 authentication failures
191
- **Network Issues**: Timeout and connectivity problems  
192
- **API Errors**: Xero API server errors or maintenance
193
- **Configuration Errors**: Invalid tenant ID or malformed parameters
194

195
### Configuration Validation Errors
196

197
The `_validate_and_transform` method validates:
198

199
- **Date Format**: `start_date` must be valid ISO 8601 format
200
- **Required Fields**: All mandatory configuration fields must be present
201
- **Type Validation**: Ensures proper data types for all configuration values
202

203
## Integration Notes
204

205
### Airbyte Platform Integration
206

207
This source connector integrates with the Airbyte platform through:
208

209
- **Standard Interface**: Implements `AbstractSource` for platform compatibility
210
- **Logging**: Uses Airbyte's structured logging for monitoring and debugging
211
- **Configuration Schema**: Follows Airbyte's configuration specification format
212
- **Stream Discovery**: Returns standard Airbyte stream objects for platform processing
213

214
### Xero API Requirements
215

216
- **Custom Connections**: Production use requires Xero Custom Connections subscription
217
- **OAuth2 Scopes**: Requires appropriate read scopes for financial data
218
- **Tenant Access**: Must have access to the specified Xero organization
219
- **Rate Limits**: Subject to Xero's API rate limiting policies