Airbyte Source Outreach

from source_outreach import SourceOutreach
from source_outreach.run import run

from source_outreach.components import CustomExtractor, CustomIncrementalSync

# Display connector specification
source-outreach spec

# Test connection with configuration
source-outreach check --config config.json

# Discover available data streams
source-outreach discover --config config.json

# Extract data from Outreach
source-outreach read --config config.json --catalog catalog.json

from source_outreach import SourceOutreach
from airbyte_cdk.entrypoint import launch
import sys

# Create and run the connector
source = SourceOutreach()
launch(source, sys.argv[1:])

{
  "client_id": "your_oauth_client_id",
  "client_secret": "your_oauth_client_secret", 
  "refresh_token": "your_oauth_refresh_token",
  "redirect_uri": "https://your-app.com/oauth/callback",
  "start_date": "2020-11-16T00:00:00Z"
}

class SourceOutreach(YamlDeclarativeSource):
    """
    Declarative source connector for Outreach API data extraction.
    
    Inherits from YamlDeclarativeSource and loads configuration from manifest.yaml.
    """
    
    def __init__(self):
        """Initialize connector with manifest.yaml configuration."""

def run():
    """
    Entry point function that creates SourceOutreach instance and launches it.
    
    Uses sys.argv[1:] for command line arguments and airbyte_cdk.entrypoint.launch
    for connector execution.
    """

class CustomExtractor(RecordExtractor):
    """
    Custom record extractor for Outreach API responses.
    
    Transforms API responses by flattening relationship data and extracting
    records from the standard Outreach API format.
    """
    
    def extract_records(self, response: requests.Response, **kwargs) -> List[Mapping[str, Any]]:
        """
        Extract and transform records from Outreach API response.
        
        Parameters:
        - response: requests.Response object from Outreach API
        - **kwargs: Additional extraction parameters
        
        Returns:
        List of transformed record dictionaries with flattened relationships
        """

class CustomIncrementalSync(DatetimeBasedCursor):
    """
    Custom incremental sync cursor for Outreach streams.
    
    Implements cursor-based incremental synchronization using updatedAt timestamps
    with Outreach-specific filtering parameters.
    """
    
    def get_request_params(
        self,
        *,
        stream_state: Optional[StreamState] = None,
        stream_slice: Optional[StreamSlice] = None,
        next_page_token: Optional[Mapping[str, Any]] = None,
    ) -> MutableMapping[str, Any]:
        """
        Generate request parameters for incremental sync.
        
        Parameters:
        - stream_state: Current stream state with cursor values
        - stream_slice: Stream slice configuration
        - next_page_token: Pagination token
        
        Returns:
        Dictionary of request parameters including filter[updatedAt] for incremental sync
        """

# Stream configuration attributes
PRIMARY_KEY = "id"  # Integer primary key
CURSOR_FIELD = "updatedAt"  # Datetime cursor for incremental sync
SYNC_MODES = ["full_refresh", "incremental"]  # Supported sync modes
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"  # ISO datetime format
BASE_URL = "https://api.outreach.io/api/v2/"  # API base URL

# OAuth configuration (defined in manifest.yaml)
OAUTH_CONFIG = {
    "client_id": "config['client_id']",
    "client_secret": "config['client_secret']", 
    "refresh_token": "config['refresh_token']",
    "redirect_uri": "config['redirect_uri']",
    "token_refresh_endpoint": "https://api.outreach.io/oauth/token",
    "grant_type": "refresh_token"
}

DEFAULT_PARAMS = {
    "count": "false",           # Disable response count
    "sort": "updatedAt",        # Sort by update timestamp
    "page[size]": "1000"        # Page size for pagination
}

# Incremental sync filter format
INCREMENTAL_FILTER = "filter[updatedAt]={cursor_value}..inf"

from typing import Any, Dict, List, Mapping, MutableMapping, Optional
from airbyte_cdk.sources.declarative.types import StreamSlice, StreamState
import requests

# Type aliases for connector usage
ConfigDict = Dict[str, Any]
RecordDict = Mapping[str, Any] 
RecordList = List[RecordDict]
RequestParams = MutableMapping[str, Any]

import subprocess
import json

# Discover available streams
result = subprocess.run([
    "source-outreach", "discover", 
    "--config", "config.json"
], capture_output=True, text=True)

catalog = json.loads(result.stdout)
streams = [stream["name"] for stream in catalog["streams"]]
print(f"Available streams: {streams}")

from source_outreach import SourceOutreach
from airbyte_cdk.models import AirbyteMessage, Type as MessageType
import json

# Configure the source
config = {
    "client_id": "your_client_id",
    "client_secret": "your_client_secret", 
    "refresh_token": "your_refresh_token",
    "redirect_uri": "https://your-app.com/oauth/callback",
    "start_date": "2024-01-01T00:00:00Z"
}

# Create connector instance
source = SourceOutreach()

# Read data (this would typically be called by Airbyte platform)
catalog = source.discover(None, config)
configured_catalog = {
    "streams": [{
        "stream": stream.json_schema,
        "sync_mode": "incremental",
        "destination_sync_mode": "append"
    } for stream in catalog.streams[:1]]  # Just first stream for example
}

# Extract records
for message in source.read(None, config, configured_catalog):
    if message.type == MessageType.RECORD:
        print(f"Record from {message.record.stream}: {message.record.data}")