How to Add a REST API Source

TL;DR

Add a new REST API data source to an OAA integration that fetches users, resources, or other entities.

Prerequisites

  • An OAA integration project set up with config.yaml

  • API endpoint documentation

  • API authentication credentials (API key, OAuth token, etc.)

  • Basic familiarity with the config.yaml structure

Steps

Step 1: Create a Connection Module

Create modules/my_api_client.py (replace my_api_client with your actual API name):

"""ThousandEyes API Client - Example from production integration"""
import requests
import logging

logger = logging.getLogger(__name__)


class MyAPIClient:
    """Custom API client for connecting to REST API"""
    
    def __init__(self, base_url=None, api_key=None, **kwargs):
        """
        Initialize API client.
        
        Args:
            base_url: Base URL of the API
            api_key: API authentication key
        """
        self.base_url = base_url or kwargs.get('my_api_base_url')
        self.api_key = api_key
        self.session = requests.Session()
        
        if self.api_key:
            self.session.headers.update({
                'Authorization': f'Bearer {self.api_key}'
            })
    
    def fetch(self, resource, **params):
        """
        Fetch data from API endpoint.
        
        Args:
            resource: API endpoint path (e.g., '/users', '/v1/resources')
            **params: Query parameters
            
        Yields:
            Individual records from the API response
        """
        url = f"{self.base_url}{resource}"
        
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            
            data = response.json()
            
            # Extract items from response (adjust based on your API structure)
            items = data.get('items', data.get('results', data.get('data', [])))
            
            logger.info(f"Fetched {len(items)} records from {resource}")
            
            yield from items
            
        except requests.RequestException as e:
            logger.error(f"API request failed: {e}")
            raise


def fetch(connection=None, source=None, **kwargs):
    """
    Fetch the data. This is the function that is called by oaa-runner.
    
    Args:
        connection: Connection configuration from config.yaml
        source: Source configuration from config.yaml
        **kwargs: Additional keyword arguments
        
    Returns:
        PETL table with fetched records (using lazy_dict_table for memory efficiency)
    """
    from oaa.modules.lazy_dicts import lazy_dict_table
    
    api = MyAPIClient(connection=connection, source=source)
    resource = source.params.get('resource', '')
    records = api.fetch(resource=resource)
    return lazy_dict_table(records)

Step 2: Update config.yaml - Add Connection

In the connections: section, add your API connection:

connections:
  my_api:
    connection_type: custom
    custom_module: my_api_client
    params:
      my_api_base_url: https://api.example.com
      api_key: ${API_KEY}  # Use environment variables for secrets

Step 3: Update config.yaml - Add Source

In the sources: section, add a source that uses your API connection:

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /users
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email
      name:
        destination_field: name
       status:
         destination_field: account_status
         custom_type: STRING  # See reference for all custom_type options

Key elements:

  • source_type: custom - Tells OAA to use a custom connection

  • connection: my_api - References the connection defined above

  • params: - Passed to your fetch() method

  • field_mapping: - Maps API response fields to OAA fields

  • custom_type: - Specifies data type for custom fields (see What are Veza Custom Properties? for all options)

Step 4: Store Credentials

Create or update .env file in your integration directory:

# .env
API_KEY=your_api_key_here

Update your config.yaml to reference environment variables:

connections:
  my_api:
    connection_type: custom
    custom_module: my_api_client
    params:
      my_api_base_url: https://api.example.com
      api_key: ${API_KEY}  # Reads from environment

Step 5: Validate Configuration

Test that your configuration is valid:

cd your_integration_folder
uv tool run --from oaa-runner oaa app validate --config=config.yaml

You should see output indicating successful validation.

Step 6: Run the Integration

Fetch data to verify everything works:

uv tool run --from oaa-runner oaa app run --config=config.yaml --save-json

Check the generated output.json to verify data is being fetched and transformed correctly.

Examples

Real Integration: ThousandEyes

From netapp/thousandeyes/config.yaml:

connections:
  api:
    connection_type: custom
    custom_module: thousandeyes
    params:
      thousandeyes_base_uri: https://api.thousandeyes.com/v7

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: api
    params:
      resource: users
    field_mapping:
      email:
        destination_field: email
      name:
        destination_field: name
      uid:
        destination_field: unique_id

From netapp/thousandeyes/modules/thousandeyes.py:

class ThousandEyesAPI:
    def __init__(self, thousandeyes_base_uri=None, **kwargs):
        self.base_uri = thousandeyes_base_uri
        self.session = requests.Session()
        self.api_token = kwargs.get('api_token')
        
        if self.api_token:
            self.session.headers.update({
                'Authorization': f'Bearer {self.api_token}'
            })
    
    def fetch(self, resource, **params):
        url = f"{self.base_uri}/{resource}"
        response = self.session.get(url, params=params)
        data = response.json()
        yield from data.get('results', [])

Real Integration: Abnormal Security

From gxo/abnormal/config.yaml:

connections:
  abnormal:
    connection_type: custom
    custom_module: abnormal
    params:
      ABNORMAL_API_BASE_URL: https://eu.rest.abnormalsecurity.com

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: abnormal
    params:
      resource: /v1/users
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email

Next Steps