Learning the OAA Runner Framework: Part 9 — Defending Against Bad Data

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

What you’ll learn: Write hooks that guard against common data quality issues — null fields, empty strings, and circular manager chains — without crashing the integration run. The BongaWonga HRIS API includes these issues intentionally as training data.

The recommended pattern is: logger.warning() then skip. Data quality problems in real integrations should produce warnings in the log, not errors that halt the run.

Prerequisites

• Completed or read Part 6 — Multiple Systems, Multiple Destinations, specifically the HRIS config section which covers destination_type: HRIS, the Employee entity, and the hris_bw.py module. This example uses those same patterns.

• An API key for the BongaWonga HRIS API (BW_HRIS_API_KEY). Get it from https://hris.bongawonga.com/profile. The profile page looks like this:

Setup

Create your working directory:

mkdir -p 09-data-quality/modules
cd 09-data-quality

Set your API key:

export BW_HRIS_API_KEY=your_hris_api_key

Copy the following code into a new file at modules/hris_bw.py (if you already created this in Part 6, you can copy it from there instead — the module is identical):

import logging
import time

from oaa.modules.base_url_session import BaseURLSession
from oaa.modules.lazy_dicts import lazy_dict_table
from oaa.modules.params_or_env import params_or_env

logger = logging.getLogger(__name__)


class HrisAPI:
    def __init__(self, connection, source):
        self.connection = connection
        self.source = source
        self.base_url = params_or_env(connection, source, 'base_url')
        self.api_key = params_or_env(connection, source, 'BW_HRIS_API_KEY')
        self._session = None

    @property
    def session(self):
        if not self._session:
            self._session = BaseURLSession(base_url=self.base_url)
            self._session.headers.update({
                'Authorization': f'Bearer {self.api_key}',
            })
        return self._session

    def fetch_paginated(self, path, resource_key='data', per_page=100):
        page = 1
        while True:
            params = {'page': page, 'per_page': per_page}
            logger.debug('Fetching %s page=%d', path, page)
            response = self.session.get(path, params=params, timeout=30)

            if response.status_code == 429:
                logger.warning('Rate limited, retrying in 5s')
                time.sleep(5)
                continue

            response.raise_for_status()
            data = response.json()
            records = data.get(resource_key, [])
            if not records:
                break
            yield from records
            pagination = data.get('pagination', {})
            if not pagination.get('has_next', False):
                break
            page += 1


def fetch(connection=None, source=None, **kwargs):
    api = HrisAPI(connection, source)
    resource_path = source.params.get('resource', '')
    resource_key = source.params.get('resource_key', 'data')
    logger.debug('Fetching from HRIS resource: %s', resource_path)
    records = api.fetch_paginated(resource_path, resource_key=resource_key)
    return lazy_dict_table(records)

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

import logging

from oaa.hooks.decorators import record_transform_non_stream

logger = logging.getLogger(__name__)


@record_transform_non_stream(target='employees')
def guard_null_email(record, provider=None, **kwargs):
    """Log a warning and skip processing for employees with no email."""
    if not getattr(record, 'email', None):
        logger.warning('Employee %s has no email — skipping', record.unique_id)
        return


@record_transform_non_stream(target='employees')
def guard_empty_external_id(record, provider=None, **kwargs):
    """Log a warning for employees with an empty external_user_id.

    `if not value` catches both None and empty string.
    """
    ext_id = getattr(record, 'external_user_id', None)
    if not ext_id:
        logger.warning(
            'Employee %s has no external_user_id — downstream linking will skip this record',
            record.unique_id,
        )


@record_transform_non_stream(target='employees')
def resolve_manager_chain(record, provider=None, **kwargs):
    """Detect and break circular manager chains using a visited set."""
    visited = set()
    current_id = getattr(record, 'manager_id', None)
    while current_id:
        if current_id in visited:
            logger.warning(
                'Circular manager chain detected at employee %s — breaking', current_id
            )
            break
        visited.add(current_id)
        manager = provider.employees.get(current_id)
        current_id = getattr(manager, 'manager_id', None) if manager else None

Create an empty config.yaml:

touch config.yaml

---

Context: intentional data quality issues in the BongaWonga HRIS API

The BongaWonga HRIS API includes three categories of data quality issue by design:

1. Null email — approximately 1% of employee records have null for the email field.

2. Empty external_user_id — approximately 3% of records have an empty string "" for external_user_id. Empty strings pass a null check but break downstream linking.

3. Circular manager chains — some records form a cycle (A.manager_id → B, B.manager_id → A). This is documented in the BongaWonga OpenAPI spec.

These are realistic problems. Production HRIS systems routinely have incomplete records, vendor-specific quirks, and relationship data that doesn’t form a clean tree.

---

Step 1: Guard against null fields

The first hook in modules/quality_hooks.py logs a warning when email is null:

@record_transform_non_stream(target='employees')
def guard_null_email(record, provider=None, **kwargs):
    if not getattr(record, 'email', None):
        logger.warning('Employee %s has no email — skipping', record.unique_id)
        return

getattr(record, 'email', None) safely reads the field even if it wasn’t mapped. if not value is True when the value is None — which covers the null case.

Important — transform ordering: record_transforms_non_stream hooks run after field mapping transforms have already executed. If the email field mapping includes a lowercase transform, that transform runs first — and crashes with AttributeError: 'NoneType' object has no attribute 'lower' before this hook ever runs. To prevent the crash, add a jinja_render step before lowercase in the field mapping that coerces null to an empty string:

      email:
        destination_field: email
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.email or '' }}"
          - name: lowercase

The transform converts null to "". The hook then sees an empty string, and if not getattr(record, 'email', None) is True for both None and "" — so it correctly warns and skips the record. Both pieces are needed: the transform prevents the crash, the hook handles the empty-email case.

Why warn and return, not raise? If this hook raised an exception, the entire run would fail whenever a single employee has no email. In practice, a missing email on one record shouldn’t block the other 9,999 records from loading. Log the issue so it’s visible in your monitoring, then skip the problematic record and continue.

What return does here: Returning None from a record_transforms_non_stream hook signals oaa-runner to skip this record entirely — it stops executing any remaining hooks for the record AND does not write the record to the destination. The employee is excluded from the push payload as if it were never loaded.

The hook is wired to run after each employee record is loaded. oaa-runner has already mapped the source fields at this point, so record.email holds the mapped OAA value.

---

Step 2: Guard against empty strings

null and "" are both falsy in Python. The same if not value check handles both:

@record_transform_non_stream(target='employees')
def guard_empty_external_id(record, provider=None, **kwargs):
    ext_id = getattr(record, 'external_user_id', None)
    if not ext_id:
        logger.warning(
            'Employee %s has no external_user_id — downstream linking will skip this record',
            record.unique_id,
        )

external_user_id is used in Part 10 to link HRIS employees to app users. An empty string would look like a valid ID but would never match anything, producing silent failures in the linking step. The warning here makes those silent failures visible.

Pattern: For any field used as a join key, check both null and empty string before the linking step. if not value covers both cases in a single expression.

---

Step 3: Detect circular manager chains

Some HRIS systems allow circular manager_id chains. Naively following a chain of manager lookups would loop forever. Guard against this with a visited set:

@record_transform_non_stream(target='employees')
def resolve_manager_chain(record, provider=None, **kwargs):
    visited = set()
    current_id = getattr(record, 'manager_id', None)
    while current_id:
        if current_id in visited:
            logger.warning(
                'Circular manager chain detected at employee %s — breaking', current_id
            )
            break
        visited.add(current_id)
        manager = provider.employees.get(current_id)
        current_id = getattr(manager, 'manager_id', None) if manager else None

provider.employees is the dict of all loaded Employee OAA objects, keyed by unique_id. The hook starts at record.manager_id and follows the chain upward. Each visited ID goes into the set. If an ID appears a second time, the chain is circular — log and break.

Why use provider.employees here? The hook has access to all loaded records via the provider object. This allows cross-record lookups without additional API calls.

Note on ordering: Employee records must be fully loaded before this hook can traverse the manager chain. oaa-runner loads all records for a source before running record_transforms_non_stream hooks, so all employees are available when any individual hook runs.

---

Wire the hooks into config

In the config header, add logic_modules:

module_directory: ./modules
logic_modules:
  - quality_hooks

In the employees source, add record_transforms_non_stream and extend the field mappings to include external_user_id and manager_id (needed by the hooks):

  employees:
    destination: Employee
    source_type: custom
    connection: bw_hris
    params:
      resource: /employees
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      # ... other fields ...
      email:
        destination_field: email
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.email or '' }}"
          - name: lowercase
      external_user_id:
        destination_field: external_user_id
        custom_type: STRING
      manager_id:
        destination_field: manager_id
        custom_type: STRING
    record_transforms_non_stream:
      - guard_null_email
      - guard_empty_external_id
      - resolve_manager_chain

external_user_id and manager_id are mapped as STRING custom properties. They don’t map to built-in OAA Employee fields, but the hooks read them via getattr(record, ...), so they need to be present in the field mapping to be available on the record object.

Also note the updated is_active mapping. Example 06’s HRIS config only handles "active" and everything else. The BongaWonga HRIS API also returns "on_leave" for employees on leave. Decide whether on_leave employees are active in your environment:

      is_active_status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.employment_status in ('active', 'on_leave') else 'false' }}"
          - name: validate_boolean

This marks on_leave employees as active. Adjust to 'false' if your policy treats on-leave employees as inactive.

---

Complete config

version: v2
log_level: INFO
destination_type: HRIS

app:
  name: BongaWonga HRIS (Data Quality)
  hris_type: bongawonga
  url: https://hris.bongawonga.com

module_directory: ./modules
logic_modules:
  - quality_hooks

connections:
  bw_hris:
    connection_type: custom
    custom_module: hris_bw
    required_env:
      - BW_HRIS_API_KEY
    params:
      base_url: https://hris.bongawonga.com/api/v1

sources:
  employees:
    destination: Employee
    source_type: custom
    connection: bw_hris
    params:
      resource: /employees
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      employee_number:
        destination_field: employee_number
      full_name:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.first_name }} {{ row.last_name }}"
      first_name:
        destination_field: first_name
      last_name:
        destination_field: last_name
      email:
        destination_field: email
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.email or '' }}"
          - name: lowercase
      employment_status:
        destination_field: employment_status
      is_active_status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.employment_status in ('active', 'on_leave') else 'false' }}"
          - name: validate_boolean
      hire_date:
        destination_field: hire_date
        transforms:
          - name: date_convert
      external_user_id:
        destination_field: external_user_id
        custom_type: STRING
      manager_id:
        destination_field: manager_id
        custom_type: STRING
    record_transforms_non_stream:
      - guard_null_email
      - guard_empty_external_id
      - resolve_manager_chain

---

Run it

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

Watch the log output. You should see warning lines similar to:

WARNING Employee emp-0042 has no email — skipping
WARNING Employee emp-0117 has no external_user_id — downstream linking will skip this record
WARNING Circular manager chain detected at employee emp-0503 — breaking

If you see no warnings, increase log_level to DEBUG to confirm the hooks are running.

---

Summary: data quality guard patterns

Issue	Check	Pattern
Null field	if not getattr(record, 'field', None)	Warn + return
Empty string	if not value (same check)	Warn + return
Circular reference	visited set in a while loop	Warn + break

---

What’s next

• Part 10 — Linking Identities Across Systems: use external_user_id to join app users with HRIS employees.

See also

• How to use record transform hooks

• Part 6 — Multiple Systems, Multiple Destinations