0
# Full Refresh Data Streams
1
2
Seven full-refresh synchronization streams that provide complete dataset snapshots on each sync. These streams handle reference data and configuration settings that require complete accuracy and are typically smaller in size.
3
4
## Capabilities
5
6
### Base Stream Class
7
8
The foundation class for all Xero streams, including full-refresh functionality.
9
10
```python { .api }
11
class XeroStream(HttpStream, ABC):
12
"""
13
Abstract base class for all Xero data streams.
14
15
Provides common functionality for HTTP requests, pagination,
16
response parsing, and Xero API integration patterns.
17
"""
18
19
url_base: str = "https://api.xero.com/api.xro/2.0/"
20
page_size: int = 100
21
current_page: int = 1
22
pagination: bool = False # Disabled by default
23
24
def __init__(self, tenant_id: str, **kwargs):
25
"""
26
Initialize Xero stream with tenant identification.
27
28
Parameters:
29
- tenant_id: Xero organization tenant ID for API requests
30
- **kwargs: Additional arguments passed to parent HttpStream
31
"""
32
33
def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
34
"""
35
Extract pagination token from API response.
36
37
Parameters:
38
- response: HTTP response from Xero API
39
40
Returns:
41
Pagination token mapping or None if no more pages
42
"""
43
44
def request_params(self, stream_state, stream_slice=None, next_page_token=None) -> MutableMapping[str, Any]:
45
"""
46
Build query parameters for API requests.
47
48
Parameters:
49
- stream_state: Current sync state (unused for full-refresh)
50
- stream_slice: Stream partition (unused)
51
- next_page_token: Pagination continuation token
52
53
Returns:
54
Query parameters mapping with page number and optional filters
55
"""
56
57
def request_headers(self, stream_state, stream_slice=None, next_page_token=None) -> Mapping[str, Any]:
58
"""
59
Build request headers for Xero API calls.
60
61
Parameters:
62
- stream_state: Current sync state (unused for full-refresh)
63
- stream_slice: Stream partition (unused)
64
- next_page_token: Pagination token (unused)
65
66
Returns:
67
Headers mapping with Xero-Tenant-Id and content type
68
"""
69
70
def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
71
"""
72
Parse JSON response and extract data records.
73
74
Parameters:
75
- response: HTTP response from Xero API
76
- **kwargs: Additional parsing arguments
77
78
Returns:
79
Iterable of record mappings with converted date formats
80
"""
81
82
def path(self, **kwargs) -> str:
83
"""
84
Generate API endpoint path from stream class name.
85
86
Returns:
87
API endpoint path (e.g., "Contacts", "Accounts")
88
"""
89
90
def data_field(self, **kwargs) -> str:
91
"""
92
Get data field name from JSON response.
93
94
Returns:
95
Data field name matching stream class name
96
"""
97
```
98
99
### Reference Data Streams
100
101
Core reference data that defines organizational structure and configuration.
102
103
#### Organizations
104
105
```python { .api }
106
class Organisations(XeroStream):
107
"""
108
Organization details and configuration settings.
109
110
Primary Key: OrganisationID
111
Pagination: Disabled (single organization per tenant)
112
Endpoint Override: "Organisation" (singular)
113
"""
114
115
primary_key = "OrganisationID"
116
117
def path(self, **kwargs) -> str:
118
"""Override to use singular 'Organisation' endpoint."""
119
return "Organisation"
120
```
121
122
#### Currencies
123
124
```python { .api }
125
class Currencies(XeroStream):
126
"""
127
Available currencies for multi-currency accounting.
128
129
Primary Key: Code (ISO currency code)
130
Pagination: Disabled (limited currency list)
131
"""
132
133
primary_key = "Code"
134
```
135
136
#### Tax Rates
137
138
```python { .api }
139
class TaxRates(XeroStream):
140
"""
141
Tax rate definitions for invoice calculations.
142
143
Primary Key: Name (tax rate name)
144
Pagination: Disabled (limited tax rates per jurisdiction)
145
"""
146
147
primary_key = "Name"
148
```
149
150
### Customization and Branding Streams
151
152
Visual and organizational customization settings.
153
154
#### Branding Themes
155
156
```python { .api }
157
class BrandingThemes(XeroStream):
158
"""
159
Custom branding themes for invoices and documents.
160
161
Primary Key: BrandingThemeID
162
Pagination: Disabled (limited branding themes)
163
"""
164
165
primary_key = "BrandingThemeID"
166
```
167
168
#### Contact Groups
169
170
```python { .api }
171
class ContactGroups(XeroStream):
172
"""
173
Contact categorization groups for customer segmentation.
174
175
Primary Key: ContactGroupID
176
Pagination: Disabled (limited contact groups)
177
"""
178
179
primary_key = "ContactGroupID"
180
```
181
182
#### Tracking Categories
183
184
```python { .api }
185
class TrackingCategories(XeroStream):
186
"""
187
Custom tracking categories for detailed financial reporting.
188
189
Primary Key: TrackingCategoryID
190
Pagination: Disabled (limited tracking categories)
191
"""
192
193
primary_key = "TrackingCategoryID"
194
```
195
196
### Template Data Streams
197
198
Recurring transaction templates and patterns.
199
200
#### Repeating Invoices
201
202
```python { .api }
203
class RepeatingInvoices(XeroStream):
204
"""
205
Recurring invoice templates for subscription billing.
206
207
Primary Key: RepeatingInvoiceID
208
Pagination: Disabled (limited recurring templates)
209
"""
210
211
primary_key = "RepeatingInvoiceID"
212
```
213
214
## Usage Examples
215
216
### Reading Full-Refresh Stream
217
218
```python
219
from source_xero.streams import Currencies
220
from airbyte_cdk.models import SyncMode
221
222
# Initialize stream
223
stream = Currencies(
224
tenant_id="your-tenant-id",
225
authenticator=authenticator
226
)
227
228
# Read all records (full dataset)
229
records = []
230
for record in stream.read_records(sync_mode=SyncMode.full_refresh):
231
records.append(record)
232
233
print(f"Found {len(records)} currencies")
234
235
# Example output: GBP, USD, EUR, AUD, NZD, etc.
236
for record in records:
237
print(f"Currency: {record['Code']} - {record['Description']}")
238
```
239
240
### Organization Information
241
242
```python
243
from source_xero.streams import Organisations
244
245
# Get organization details
246
stream = Organisations(
247
tenant_id="your-tenant-id",
248
authenticator=authenticator
249
)
250
251
# Should return exactly one organization record
252
org_records = list(stream.read_records(sync_mode=SyncMode.full_refresh))
253
organization = org_records[0]
254
255
print(f"Organization: {organization['Name']}")
256
print(f"Base Currency: {organization['BaseCurrency']}")
257
print(f"Country: {organization['CountryCode']}")
258
print(f"Tax Number: {organization['TaxNumber']}")
259
```
260
261
### Tax Configuration
262
263
```python
264
from source_xero.streams import TaxRates
265
266
# Get all tax rates
267
stream = TaxRates(
268
tenant_id="your-tenant-id",
269
authenticator=authenticator
270
)
271
272
tax_rates = list(stream.read_records(sync_mode=SyncMode.full_refresh))
273
274
# Display tax configuration
275
for tax_rate in tax_rates:
276
print(f"Tax: {tax_rate['Name']} - {tax_rate['TaxType']} ({tax_rate['EffectiveRate']}%)")
277
```
278
279
### Branding and Customization
280
281
```python
282
from source_xero.streams import BrandingThemes, ContactGroups
283
284
# Get branding themes
285
branding_stream = BrandingThemes(
286
tenant_id="your-tenant-id",
287
authenticator=authenticator
288
)
289
290
themes = list(branding_stream.read_records(sync_mode=SyncMode.full_refresh))
291
for theme in themes:
292
print(f"Theme: {theme['Name']} (ID: {theme['BrandingThemeID']})")
293
294
# Get contact groups
295
contact_groups_stream = ContactGroups(
296
tenant_id="your-tenant-id",
297
authenticator=authenticator
298
)
299
300
groups = list(contact_groups_stream.read_records(sync_mode=SyncMode.full_refresh))
301
for group in groups:
302
print(f"Contact Group: {group['Name']} ({len(group.get('Contacts', []))} contacts)")
303
```
304
305
## Data Characteristics
306
307
### Dataset Sizes
308
309
Full-refresh streams typically handle smaller, reference datasets:
310
311
```python
312
TypicalDatasetSizes = {
313
"Organisations": 1, # Single organization per tenant
314
"Currencies": "10-50", # Available currencies
315
"TaxRates": "5-20", # Tax rates per jurisdiction
316
"BrandingThemes": "1-10", # Custom branding themes
317
"ContactGroups": "0-50", # Customer segmentation groups
318
"TrackingCategories": "0-20", # Custom tracking categories
319
"RepeatingInvoices": "0-100" # Recurring invoice templates
320
}
321
```
322
323
### Update Frequency
324
325
Reference data changes infrequently, making full-refresh appropriate:
326
327
- **Organizations**: Rarely changes (company details, settings)
328
- **Currencies**: Occasionally updated (new currency support)
329
- **Tax Rates**: Periodic updates (tax law changes)
330
- **Branding**: Occasional customization updates
331
- **Contact Groups**: Infrequent organizational changes
332
- **Tracking Categories**: Rare structural changes
333
- **Repeating Invoices**: Occasional template modifications
334
335
### Data Dependencies
336
337
Many incremental streams reference full-refresh data:
338
339
```python
340
DataDependencies = {
341
"Invoices": ["Currencies", "TaxRates", "BrandingThemes"],
342
"Contacts": ["ContactGroups", "Currencies"],
343
"Accounts": ["TaxRates", "Currencies"],
344
"Items": ["TaxRates"],
345
"BankTransactions": ["Currencies"]
346
}
347
```
348
349
## Performance Characteristics
350
351
### No Pagination Required
352
353
Most full-refresh streams disable pagination due to small dataset sizes:
354
355
```python
356
# Pagination disabled streams
357
NoPaginationStreams = [
358
"Organisations", # Single record
359
"Currencies", # Small reference list
360
"TaxRates", # Small configuration set
361
"BrandingThemes", # Limited customization options
362
"ContactGroups", # Small organizational structure
363
"TrackingCategories", # Limited custom categories
364
"RepeatingInvoices" # Manageable template count
365
]
366
```
367
368
### Fast Synchronization
369
370
Full-refresh streams sync quickly due to:
371
372
- **Small Data Volume**: Typically under 1000 records total
373
- **Simple Structure**: Minimal nested objects or relationships
374
- **No State Management**: No cursor tracking or incremental logic
375
- **Single API Call**: Most streams complete in one request
376
377
### Memory Efficiency
378
379
Full-refresh streams are memory-efficient:
380
381
```python
382
# Typical memory usage per stream:
383
MemoryUsage = {
384
"Organisations": "< 1KB",
385
"Currencies": "< 10KB",
386
"TaxRates": "< 5KB",
387
"BrandingThemes": "< 20KB",
388
"ContactGroups": "< 15KB",
389
"TrackingCategories": "< 10KB",
390
"RepeatingInvoices": "< 50KB"
391
}
392
```
393
394
## Error Handling
395
396
### Connection Failures
397
398
Full-refresh streams handle API connectivity issues:
399
400
- **Network Timeouts**: Automatic retry with exponential backoff
401
- **Authentication Errors**: Token refresh and request retry
402
- **Server Errors**: Retry logic for 500+ status codes
403
404
### Data Validation
405
406
Reference data undergoes validation:
407
408
```python
409
# Common validation patterns:
410
DataValidation = {
411
"Organisations": "Validates required fields (Name, BaseCurrency)",
412
"Currencies": "Validates ISO currency codes and exchange rates",
413
"TaxRates": "Validates tax percentages and effective dates",
414
"BrandingThemes": "Validates theme assets and configurations"
415
}
416
```
417
418
### Consistency Checks
419
420
Full-refresh ensures data consistency:
421
422
- **Complete Dataset**: Always returns full current state
423
- **No Partial Updates**: Eliminates incremental sync inconsistencies
424
- **Reference Integrity**: Maintains relationship consistency with other streams
425
- **Schema Validation**: Enforces JSON schema compliance for all records