Full Refresh Data Streams

class XeroStream(HttpStream, ABC):
    """
    Abstract base class for all Xero data streams.
    
    Provides common functionality for HTTP requests, pagination,
    response parsing, and Xero API integration patterns.
    """
    
    url_base: str = "https://api.xero.com/api.xro/2.0/"
    page_size: int = 100
    current_page: int = 1
    pagination: bool = False                     # Disabled by default
    
    def __init__(self, tenant_id: str, **kwargs):
        """
        Initialize Xero stream with tenant identification.
        
        Parameters:
        - tenant_id: Xero organization tenant ID for API requests
        - **kwargs: Additional arguments passed to parent HttpStream
        """
    
    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        """
        Extract pagination token from API response.
        
        Parameters:
        - response: HTTP response from Xero API
        
        Returns:
        Pagination token mapping or None if no more pages
        """
    
    def request_params(self, stream_state, stream_slice=None, next_page_token=None) -> MutableMapping[str, Any]:
        """
        Build query parameters for API requests.
        
        Parameters:
        - stream_state: Current sync state (unused for full-refresh)
        - stream_slice: Stream partition (unused)
        - next_page_token: Pagination continuation token
        
        Returns:
        Query parameters mapping with page number and optional filters
        """
    
    def request_headers(self, stream_state, stream_slice=None, next_page_token=None) -> Mapping[str, Any]:
        """
        Build request headers for Xero API calls.
        
        Parameters:
        - stream_state: Current sync state (unused for full-refresh)
        - stream_slice: Stream partition (unused)
        - next_page_token: Pagination token (unused)
        
        Returns:
        Headers mapping with Xero-Tenant-Id and content type
        """
    
    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
        """
        Parse JSON response and extract data records.
        
        Parameters:
        - response: HTTP response from Xero API
        - **kwargs: Additional parsing arguments
        
        Returns:
        Iterable of record mappings with converted date formats
        """
    
    def path(self, **kwargs) -> str:
        """
        Generate API endpoint path from stream class name.
        
        Returns:
        API endpoint path (e.g., "Contacts", "Accounts")
        """
    
    def data_field(self, **kwargs) -> str:
        """
        Get data field name from JSON response.
        
        Returns:
        Data field name matching stream class name
        """

class Organisations(XeroStream):
    """
    Organization details and configuration settings.
    
    Primary Key: OrganisationID
    Pagination: Disabled (single organization per tenant)
    Endpoint Override: "Organisation" (singular)
    """
    
    primary_key = "OrganisationID"
    
    def path(self, **kwargs) -> str:
        """Override to use singular 'Organisation' endpoint."""
        return "Organisation"

class Currencies(XeroStream):
    """
    Available currencies for multi-currency accounting.
    
    Primary Key: Code (ISO currency code)
    Pagination: Disabled (limited currency list)
    """
    
    primary_key = "Code"

class TaxRates(XeroStream):
    """
    Tax rate definitions for invoice calculations.
    
    Primary Key: Name (tax rate name)
    Pagination: Disabled (limited tax rates per jurisdiction)
    """
    
    primary_key = "Name"

class BrandingThemes(XeroStream):
    """
    Custom branding themes for invoices and documents.
    
    Primary Key: BrandingThemeID
    Pagination: Disabled (limited branding themes)
    """
    
    primary_key = "BrandingThemeID"

class ContactGroups(XeroStream):
    """
    Contact categorization groups for customer segmentation.
    
    Primary Key: ContactGroupID
    Pagination: Disabled (limited contact groups)
    """
    
    primary_key = "ContactGroupID"

class TrackingCategories(XeroStream):
    """
    Custom tracking categories for detailed financial reporting.
    
    Primary Key: TrackingCategoryID
    Pagination: Disabled (limited tracking categories)
    """
    
    primary_key = "TrackingCategoryID"

class RepeatingInvoices(XeroStream):
    """
    Recurring invoice templates for subscription billing.
    
    Primary Key: RepeatingInvoiceID
    Pagination: Disabled (limited recurring templates)
    """
    
    primary_key = "RepeatingInvoiceID"

from source_xero.streams import Currencies
from airbyte_cdk.models import SyncMode

# Initialize stream
stream = Currencies(
    tenant_id="your-tenant-id",
    authenticator=authenticator
)

# Read all records (full dataset)
records = []
for record in stream.read_records(sync_mode=SyncMode.full_refresh):
    records.append(record)

print(f"Found {len(records)} currencies")

# Example output: GBP, USD, EUR, AUD, NZD, etc.
for record in records:
    print(f"Currency: {record['Code']} - {record['Description']}")

from source_xero.streams import Organisations

# Get organization details
stream = Organisations(
    tenant_id="your-tenant-id", 
    authenticator=authenticator
)

# Should return exactly one organization record
org_records = list(stream.read_records(sync_mode=SyncMode.full_refresh))
organization = org_records[0]

print(f"Organization: {organization['Name']}")
print(f"Base Currency: {organization['BaseCurrency']}")
print(f"Country: {organization['CountryCode']}")
print(f"Tax Number: {organization['TaxNumber']}")

from source_xero.streams import TaxRates

# Get all tax rates
stream = TaxRates(
    tenant_id="your-tenant-id",
    authenticator=authenticator
)

tax_rates = list(stream.read_records(sync_mode=SyncMode.full_refresh))

# Display tax configuration
for tax_rate in tax_rates:
    print(f"Tax: {tax_rate['Name']} - {tax_rate['TaxType']} ({tax_rate['EffectiveRate']}%)")

from source_xero.streams import BrandingThemes, ContactGroups

# Get branding themes
branding_stream = BrandingThemes(
    tenant_id="your-tenant-id",
    authenticator=authenticator
)

themes = list(branding_stream.read_records(sync_mode=SyncMode.full_refresh))
for theme in themes:
    print(f"Theme: {theme['Name']} (ID: {theme['BrandingThemeID']})")

# Get contact groups
contact_groups_stream = ContactGroups(
    tenant_id="your-tenant-id",
    authenticator=authenticator
)

groups = list(contact_groups_stream.read_records(sync_mode=SyncMode.full_refresh))
for group in groups:
    print(f"Contact Group: {group['Name']} ({len(group.get('Contacts', []))} contacts)")

TypicalDatasetSizes = {
    "Organisations": 1,           # Single organization per tenant
    "Currencies": "10-50",        # Available currencies
    "TaxRates": "5-20",          # Tax rates per jurisdiction
    "BrandingThemes": "1-10",    # Custom branding themes
    "ContactGroups": "0-50",     # Customer segmentation groups
    "TrackingCategories": "0-20", # Custom tracking categories
    "RepeatingInvoices": "0-100" # Recurring invoice templates
}

DataDependencies = {
    "Invoices": ["Currencies", "TaxRates", "BrandingThemes"],
    "Contacts": ["ContactGroups", "Currencies"],
    "Accounts": ["TaxRates", "Currencies"],
    "Items": ["TaxRates"],
    "BankTransactions": ["Currencies"]
}

# Pagination disabled streams
NoPaginationStreams = [
    "Organisations",      # Single record
    "Currencies",         # Small reference list
    "TaxRates",          # Small configuration set
    "BrandingThemes",     # Limited customization options
    "ContactGroups",      # Small organizational structure
    "TrackingCategories", # Limited custom categories
    "RepeatingInvoices"   # Manageable template count
]

# Typical memory usage per stream:
MemoryUsage = {
    "Organisations": "< 1KB",
    "Currencies": "< 10KB", 
    "TaxRates": "< 5KB",
    "BrandingThemes": "< 20KB",
    "ContactGroups": "< 15KB",
    "TrackingCategories": "< 10KB",
    "RepeatingInvoices": "< 50KB"
}

# Common validation patterns:
DataValidation = {
    "Organisations": "Validates required fields (Name, BaseCurrency)",
    "Currencies": "Validates ISO currency codes and exchange rates",
    "TaxRates": "Validates tax percentages and effective dates",
    "BrandingThemes": "Validates theme assets and configurations"
}