Learning the OAA Runner Framework: Part 7 — Fetching CSV Data over HTTP

This is Part 7 of the Learning the OAA Runner Framework series. It builds on Part 6 — Multiple Systems, Multiple Destinations.

What you’ll learn: Fetch CSV data delivered over HTTP, parse it in a custom module using io.StringIO, and map CSV columns to OAA fields. This is an uncommon but real pattern — some older SaaS tools export data as CSV over HTTP rather than JSON. The BongaWonga CustomApps API is one such example.

Prerequisites

• Completed or read Part 5 — Connecting to a REST API and Part 6 — Multiple Systems, Multiple Destinations.

• An API key for the BongaWonga CustomApps API (BW_CUSTOMAPPS_API_KEY). Get it from https://customapps.bongawonga.com/profile.

Setup

Create your working directory:

mkdir -p 07-csv-api/modules
cd 07-csv-api

Set your API key:

export BW_CUSTOMAPPS_API_KEY=your_api_key

Copy the following code into a new file at modules/customapps_bw.py:

import io
import logging

import petl
import requests

from oaa.modules.lazy_dicts import lazy_dict_table

logger = logging.getLogger(__name__)


class CustomAppsAPI:
    def __init__(self, connection, source):
        import os
        self.base_url = connection.params.get('base_url', 'https://customapps.bongawonga.com')
        self.api_key = os.environ.get('BW_CUSTOMAPPS_API_KEY', '')

    def fetch_csv(self, app_id):
        url = f"{self.base_url}/api/v1/apps/{app_id}/csv"
        resp = requests.get(url, headers={'Authorization': f'Bearer {self.api_key}'}, timeout=30)
        if resp.status_code == 401:
            logger.warning('Bearer auth failed, retrying with X-API-Key header')
            resp = requests.get(url, headers={'X-API-Key': self.api_key}, timeout=30)
        resp.raise_for_status()
        return resp.text


def fetch(connection=None, source=None, **kwargs):
    """Fetch CSV data from the BongaWonga CustomApps API.

    :param connection: Connection configuration from config.yaml
    :param source: Source configuration from config.yaml
    :returns: PETL table using lazy_dict_table for memory efficiency
    """
    api = CustomAppsAPI(connection, source)
    app_id = source.params.get('app_id', 'waypoints')
    logger.info('Fetching CSV for app_id=%s', app_id)
    csv_text = api.fetch_csv(app_id)
    table = petl.fromcsv(io.StringIO(csv_text))
    return lazy_dict_table(list(petl.dicts(table)))

Copy the following into a new file at config.yaml:

version: v2
log_level: INFO

You’ll build it step by step below.

---

Step 1: Define the app and module directory

Open config.yaml and add the app block and module directory:

app:
  name: BongaWonga CustomApps (Waypoints)
  application_type: Custom App

module_directory: ./modules

Same pattern as earlier examples.

---

Step 2: Define the connection

The BongaWonga CustomApps API authenticates with a static API key — same pattern as Parts 5–6: set once on the session, sent as a Bearer token header on each request.

Add the connection:

connections:
  bw_customapps:
    connection_type: custom
    custom_module: customapps_bw
    required_env:
      - BW_CUSTOMAPPS_API_KEY
    params:
      base_url: https://customapps.bongawonga.com

required_env declares that BW_CUSTOMAPPS_API_KEY must be present before the run starts. oaa-runner checks this before invoking your module and exits early with a clear error if the variable is missing.

API key header fallback. The BongaWonga CustomApps API prefers Authorization: Bearer <key> but falls back to X-API-Key: <key> on legacy endpoints. The fetch module handles this automatically — see Step 3.

---

Step 3: Understand the fetch module

Two module types: fetch vs hook

Parts 3–4 introduced logic modules registered in logic_modules. These run as hooks during record processing (e.g., record_transforms_non_stream).

Fetch modules work differently. They are registered inside a connection definition via custom_module, and oaa-runner calls their fetch() function once per source to pull data. They do not go in logic_modules.

Type	How to register	When it runs
Hook/logic module	logic_modules: [module_name]	During record processing
Fetch module	custom_module: module_name inside connections	At data fetch time, once per source

A module file can contain both a fetch() function (making it a fetch module for that connection) and hook decorators (making it a logic module too) — but they must be registered separately via both custom_module and logic_modules.

modules/customapps_bw.py was created in Setup. The key difference from Parts 5–6 is in how the response is parsed.

Most REST APIs return JSON. This API returns raw CSV text. The module uses io.StringIO to wrap the response text so that petl.fromcsv() can read it as a file-like object:

def fetch_csv(self, app_id):
    url = f"{self.base_url}/api/v1/apps/{app_id}/csv"
    resp = requests.get(url, headers={'Authorization': f'Bearer {self.api_key}'}, timeout=30)
    if resp.status_code == 401:
        resp = requests.get(url, headers={'X-API-Key': self.api_key}, timeout=30)
    resp.raise_for_status()
    return resp.text

def fetch(connection=None, source=None, **kwargs):
    api = CustomAppsAPI(connection, source)
    app_id = source.params.get('app_id', 'waypoints')
    csv_text = api.fetch_csv(app_id)
    table = petl.fromcsv(io.StringIO(csv_text))
    return lazy_dict_table(list(petl.dicts(table)))

io.StringIO(csv_text) creates an in-memory file object from the response string. petl.fromcsv() reads it exactly as it would a file. petl.dicts() converts each row to a dict, which lazy_dict_table() wraps for memory-efficient iteration.

Why not petl.fromdicts()? The lazy_dict_table() function wraps an iterator and evaluates lazily — data is only processed when oaa-runner reads it. petl.fromdicts() loads all rows into memory at once. For small CSV files the difference is minor; for large exports, lazy_dict_table is significantly more efficient.

The app_id is configurable via source.params. The waypoints app is the default.

---

Step 4: Add the users source

The waypoints CSV has these columns:

employee_id, email_address, first, last, full_name, status, department,
job_title, hire_date, office_location, manager_id, is_remote

Add a minimal source to verify the connection before mapping all fields:

sources:
  waypoints_users:
    destination: LocalUser
    source_type: custom
    connection: bw_customapps
    params:
      app_id: waypoints
    field_mapping:
      employee_id:
        destination_field: unique_id
      email_address:
        destination_field: email
        transforms:
          - name: lowercase
      full_name:
        destination_field: name

Note that source.params here uses app_id instead of resource and resource_key — because the CSV endpoint URL is constructed differently from the JSON endpoints in Parts 5–6. The module reads app_id directly from source.params.

app_id: waypoints is the name of the specific BongaWonga CustomApps application whose user list to export. The BongaWonga training environment hosts several apps; waypoints is the one used in this example. Change this value to target a different app.

---

Try it

Run with just the three fields mapped:

uv tool run --from oaa-runner oaa custom_app run \
  --config-file=config.yaml --no-push

You should see employee records loading. If you get a 401 error, verify your BW_CUSTOMAPPS_API_KEY is set correctly.

---

Step 5: Add the remaining field mappings

Now add the rest of the fields. Extend the field_mapping block:

    field_mapping:
      employee_id:
        destination_field: unique_id
      email_address:
        destination_field: email
        transforms:
          - name: lowercase
      full_name:
        destination_field: name
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
      hire_date:
        destination_field: hire_date
        transforms:
          - name: date_convert
      department:
        destination_field: department
        custom_type: STRING
      job_title:
        destination_field: title
        custom_type: STRING
      is_remote:
        destination_field: is_remote
        custom_type: BOOLEAN

Field notes:

• status uses jinja_render then validate_boolean — same pattern as Parts 5–6. The CSV field value "active" becomes true; anything else becomes false.

• hire_date uses date_convert. The BongaWonga CustomApps API exports dates as ISO 8601 strings (2021-03-15), which date_convert handles.

• department, job_title, is_remote are custom properties. Use custom_type to tell Veza the data type — STRING, STRING, and BOOLEAN respectively.

• office_location, manager_id, first, and last are intentionally omitted here. You can add them as STRING custom properties if needed.

---

Complete config

version: v2
log_level: INFO

app:
  name: BongaWonga CustomApps (Waypoints)
  application_type: Custom App

module_directory: ./modules

connections:
  bw_customapps:
    connection_type: custom
    custom_module: customapps_bw
    required_env:
      - BW_CUSTOMAPPS_API_KEY
    params:
      base_url: https://customapps.bongawonga.com

sources:
  waypoints_users:
    destination: LocalUser
    source_type: custom
    connection: bw_customapps
    params:
      app_id: waypoints
    field_mapping:
      employee_id:
        destination_field: unique_id
      email_address:
        destination_field: email
        transforms:
          - name: lowercase
      full_name:
        destination_field: name
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
      hire_date:
        destination_field: hire_date
        transforms:
          - name: date_convert
      department:
        destination_field: department
        custom_type: STRING
      job_title:
        destination_field: title
        custom_type: STRING
      is_remote:
        destination_field: is_remote
        custom_type: BOOLEAN

---

Run it

export BW_CUSTOMAPPS_API_KEY=your_api_key
uv tool run --from oaa-runner oaa custom_app run \
  --config-file=config.yaml --no-push

If it worked, you’ll see:

• CSV file fetched over HTTP from the API

• All employee records parsed and loaded with full field mappings

• No field mapping errors in the output

• Final output: status: true, report: []

---

What’s next

• Part 8 — Scoping Permissions to Resources: scope role assignments to specific resources instead of the whole application.

See also

• How to configure field mappings

• How to write a fetch function

• How to use record transform hooks