How To Write a Fetch Function?¶

Overview¶

The fetch() function is the entry point called by oaa-runner when a source uses a custom module. It must be defined in your custom module (e.g., modules/my_api_client.py) and is responsible for retrieving data from your source system.

Function Signature¶

Your fetch() function must accept these parameters:

def fetch(connection=None, source=None, **kwargs):
    """
    Fetch the data. This is the function that is called by oaa-runner.
    
    Args:
        connection: Connection configuration object from config.yaml
                   - Has a `params` dict with connection-specific settings
        source: Source configuration object from config.yaml
               - Has a `params` dict with source-specific settings
        **kwargs: Additional keyword arguments from the framework
        
    Returns:
        PETL table using lazy_dict_table() for memory-efficient processing
    """
    pass

Minimal Example¶

from oaa.modules.lazy_dicts import lazy_dict_table

def fetch(connection=None, source=None, **kwargs):
    """Return a lazy-evaluated PETL table"""
    def data_generator():
        yield {'id': 1, 'name': 'User 1'}
        yield {'id': 2, 'name': 'User 2'}
    
    return lazy_dict_table(data_generator())

Practical Example with API Client¶

import requests
from oaa.modules.lazy_dicts import lazy_dict_table
from dotenv import load_dotenv

load_dotenv()

class MyAPIClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {self.api_key}'
        })
    
    def fetch_records(self, resource):
        """
        Fetch records from API endpoint.
        Yields records one at a time for memory efficiency.
        """
        url = f"{self.base_url}{resource}"
        response = self.session.get(url, timeout=30)
        response.raise_for_status()
        
        data = response.json()
        items = data.get('items', data.get('data', []))
        
        # Yield one at a time to avoid loading entire dataset into memory
        yield from items


def fetch(connection=None, source=None, **kwargs):
    """
    Fetch the data. This is the function that is called by oaa-runner.
    
    Returns lazy_dict_table for memory-efficient processing of large datasets.
    """
    api = MyAPIClient(**connection.params)
    resource = source.params.get('resource', '')
    records = api.fetch_records(resource=resource)
    
    # Use lazy_dict_table to wrap the generator with lazy evaluation
    # This avoids loading all records into memory at once
    return lazy_dict_table(records)

Accessing Configuration¶

Both connection and source objects have a params dictionary containing configuration from config.yaml:

def fetch(connection=None, source=None, **kwargs):
    # From connections section in config.yaml
    base_url = connection.params['base_url']
    api_key = connection.params['api_key']
    
    # From sources section in config.yaml
    resource = source.params.get('resource', '/users')
    page_size = source.params.get('page_size', 100)
    
    # ... fetch and return data

Best Practice vs Production Reality¶

⭐ Recommended Approach (Best Practice):

from oaa.modules.lazy_dicts import lazy_dict_table
return lazy_dict_table(data_generator())  # Memory-efficient, lazy evaluation

📊 Current Production Usage (Still Valid):

import petl
return petl.fromdicts(list(data_generator()))  # Works fine for smaller datasets

Why the difference?

40+ production integrations still use petl.fromdicts() - this works well when datasets are small-to-medium (< 100K records)
3 production integrations (Wolken, Zoom, Sage Intact) use lazy_dict_table() - this is the forward-looking best practice for memory efficiency with large datasets
Both approaches are valid; choose based on your data volume

Recommendation:

Use lazy_dict_table() for new integrations
Use petl.fromdicts() only if you know your data is small or have memory constraints
See How To Handle API Pagination for memory-efficient techniques

Key Points¶

Function signature must be exactly def fetch(connection=None, source=None, **kwargs):
Import PETL operations from correct location: from oaa.modules.lazy_dicts import lazy_dict_table
The fetch() function is called once per source that uses your custom module

Data should be yielded (using generators) rather than accumulated in lists:

# ✅ Good: Uses generator - memory efficient
def fetch_records(resource):
    for item in response_data:
        yield item

# ❌ Bad: Loads all into memory at once
def fetch_records(resource):
    items = []
    for item in response_data:
        items.append(item)
    return items

Handle errors gracefully and log important events

Production Examples¶

cloud_software_group/wolken/modules/wolken.py - Uses lazy_dict_table() for memory-efficient pagination
generic/zoom/modules/zoom.py - Uses lazy_dict_table() with large user datasets
stark_technologies/sage_intact/modules/sage_intacct.py - Uses lazy_dict_table() for financial data