Learning the OAA Runner Framework: Part 4 — Writing Your First Hook

This is Part 4 of the Learning the OAA Runner Framework series. It builds on Part 3 — Modeling a Custom Application.

What you’ll learn: Write record transform hooks to build relationships that config.yaml cannot express: deriving a groups source from user rows, then assigning users to groups and roles by iterating across the fully loaded provider graph.

Before you start

Hooks have a real cost — they are Python code that must be maintained alongside the config. Before writing a hook, verify that config.yaml can’t solve the problem with transforms or generators. Read Exhaust config.yaml before writing a hook.

This example uses hooks for two reasons generators cannot handle:

1. A derived source — departments is not its own CSV but is created by scanning the users source for unique DEPARTMENT values.

2. Cross-file role assignments — roles come from a separate assignments CSV that maps user IDs to role names. A user can appear multiple times in that file.

Prerequisites

• Completed or read Part 3 — Modeling a Custom Application.

• Read Exhaust config.yaml before writing a hook.

Setup

Create your working directory:

mkdir -p 04-csv-hooks/data 04-csv-hooks/modules
cd 04-csv-hooks

Copy the following content into a new file at data/users.csv — same format as Part 3:

USER_ID,EMAIL,FIRST_NAME,LAST_NAME,IS_ACTIVE,HIRE_DATE,DEPARTMENT,ROLE
1001,alice@example.com,Alice,Anderson,yes,2020-01-15,Engineering,Admin
1002,bob@example.com,Bob,Baker,1,2019-03-22,Engineering,Developer
1003,carol@example.com,Carol,Chen,true,2021-06-01,Marketing,Viewer

Copy the following content into a new file at data/assignments.csv:

USER_ID,ROLE_NAME
1001,Admin
1001,Billing Viewer
1002,Developer
1002,Code Reviewer

Users can appear multiple times — one row per role assignment. This many-to-many relationship can’t be expressed as a generator.

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

import logging

from oaa.hooks.decorators import record_transform

logger = logging.getLogger(__name__)


@record_transform(stream=False, target='departments')
def assign_department_members(record, provider=None, **kwargs):
    user = provider.local_users.get(record.USER_ID)
    if user:
        user.add_group(record.DEPARTMENT)
        logger.debug('Added user %s to department %s', record.USER_ID, record.DEPARTMENT)


@record_transform(stream=False, target='roles')
def assign_roles(record, provider=None, **kwargs):
    user = provider.local_users.get(record.USER_ID)
    if user:
        user.add_role(record.ROLE_NAME, apply_to_application=True)
        logger.debug('Assigned role %s to user %s', record.ROLE_NAME, record.USER_ID)
    else:
        logger.warning('User %s not found for role %s', record.USER_ID, record.ROLE_NAME)

Create an empty config.yaml:

touch config.yaml

---

Step 1: Create the config skeleton

Open config.yaml. Add the required header and enable the hook module:

version: v2
log_level: INFO

app:
  name: Example App - CSV with Hooks
  application_type: Custom App

module_directory: ./modules
logic_modules:
  - app_hooks

• module_directory — tells oaa-runner where to find Python hook modules.

• logic_modules — lists which module files (without .py) to load and register.

Without both entries, hook functions in app_hooks.py are never registered and any record_transforms_non_stream entries silently fail.

---

Step 2: Add the users source

Open config.yaml and add the sources block with the users source. This is similar to Part 3, with two differences: the generators are removed (replaced by hooks below), and the name mapping uses a different template:

sources:
  users:
    destination: LocalUser
    source_type: csv
    csv_filepath: data/users.csv
    field_mapping:
      USER_ID:
        destination_field: unique_id
      EMAIL:
        destination_field: email
        transforms:
          - name: lowercase
      DISPLAY_NAME:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.EMAIL.split('@')[0] }}"
      IS_ACTIVE:
        destination_field: is_active
        transforms:
          - name: validate_boolean
      HIRE_DATE:
        destination_field: hire_date
        transforms:
          - name: date_convert
      DEPARTMENT:
        destination_field: department
        custom_type: STRING

Why the name template changed: Part 3 used {{ row.FIRST_NAME }} {{ row.LAST_NAME }}. This part uses {{ row.EMAIL.split('@')[0] }} — the local part of the email address — to show that Jinja templates can call Python string methods, not just reference fields. Both produce a valid name value; the switch is intentional to demonstrate the pattern, not because the data changed.

---

Step 3: Add a derived source for departments

Instead of a second CSV for departments, derive them from the users source. Add this after the users source:

  departments:
    destination: LocalGroup
    source_type: source
    source: users
    field_mapping:
      DEPARTMENT:
        destination_field: unique_id
      DEPARTMENT_NAME:
        destination_field: name
        source_field: DEPARTMENT
    record_transforms_non_stream:
      - assign_department_members

source_type: source scans the named source (users) and produces one row per unique value of the DEPARTMENT column. Each unique department becomes a LocalGroup record.

record_transforms_non_stream lists the hook to run after all records are loaded. The hook will assign each user to their department group.

---

Step 4: Add the roles source from the assignments CSV

  roles:
    destination: LocalRole
    source_type: csv
    csv_filepath: data/assignments.csv
    field_mapping:
      ROLE_NAME:
        destination_field: unique_id
      ROLE_NAME_DISPLAY:
        destination_field: name
        source_field: ROLE_NAME
    record_transforms_non_stream:
      - assign_roles

source_field: ROLE_NAME on the ROLE_NAME_DISPLAY entry tells oaa-runner to read from the ROLE_NAME column even though the mapping key is ROLE_NAME_DISPLAY. This is the pattern for mapping the same source column to two different destination fields.

---

Step 5: Review the hook module

modules/app_hooks.py was created in Setup. Here is what each function does:

Note on decorator syntax: This module uses @record_transform(stream=False, target='...'). The stream=False argument is what wires the function to run after all records are loaded — the same behavior triggered by record_transforms_non_stream in config.yaml. Parts 5 and later use a shorter alias, @record_transform_non_stream(target='...'), which is equivalent. Both forms work; the tutorials transition to the alias in the REST API track.

import logging

from oaa.hooks.decorators import record_transform

logger = logging.getLogger(__name__)


@record_transform(stream=False, target='departments')
def assign_department_members(record, provider=None, **kwargs):
    user = provider.local_users.get(record.USER_ID)
    if user:
        user.add_group(record.DEPARTMENT)
        logger.debug('Added user %s to department %s', record.USER_ID, record.DEPARTMENT)


@record_transform(stream=False, target='roles')
def assign_roles(record, provider=None, **kwargs):
    user = provider.local_users.get(record.USER_ID)
    if user:
        user.add_role(record.ROLE_NAME, apply_to_application=True)
        logger.debug('Assigned role %s to user %s', record.ROLE_NAME, record.USER_ID)
    else:
        logger.warning('User %s not found for role %s', record.USER_ID, record.ROLE_NAME)

Key points about this module:

@record_transform(stream=False, target='departments')

• stream=False — the hook runs after all records from all sources are loaded (record_transforms_non_stream in config). This is required when the hook needs to look up records from a different source.

• target='departments' — binds this function to the departments source. The value must exactly match a source name in config.yaml. oaa-runner calls this function once per record from that source.

provider.local_users.get(record.USER_ID)

provider.local_users is a dictionary of all loaded LocalUser objects, keyed by unique_id. Since all sources are loaded before the hook runs, this lookup always finds the user if the ID is valid.

user.add_group(record.DEPARTMENT)

Adds the user to the group whose ID is record.DEPARTMENT. The group was created when the departments source loaded.

user.add_role(record.ROLE_NAME, apply_to_application=True)

Assigns the role to the user and marks it as an application-level role assignment.

---

Complete config

version: v2
log_level: INFO

app:
  name: Example App - CSV with Hooks
  application_type: Custom App

module_directory: ./modules
logic_modules:
  - app_hooks

sources:
  users:
    destination: LocalUser
    source_type: csv
    csv_filepath: data/users.csv
    field_mapping:
      USER_ID:
        destination_field: unique_id
      EMAIL:
        destination_field: email
        transforms:
          - name: lowercase
      DISPLAY_NAME:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.EMAIL.split('@')[0] }}"
      IS_ACTIVE:
        destination_field: is_active
        transforms:
          - name: validate_boolean
      HIRE_DATE:
        destination_field: hire_date
        transforms:
          - name: date_convert
      DEPARTMENT:
        destination_field: department
        custom_type: STRING

  departments:
    destination: LocalGroup
    source_type: source
    source: users
    field_mapping:
      DEPARTMENT:
        destination_field: unique_id
      DEPARTMENT_NAME:
        destination_field: name
        source_field: DEPARTMENT
    record_transforms_non_stream:
      - assign_department_members

  roles:
    destination: LocalRole
    source_type: csv
    csv_filepath: data/assignments.csv
    field_mapping:
      ROLE_NAME:
        destination_field: unique_id
      ROLE_NAME_DISPLAY:
        destination_field: name
        source_field: ROLE_NAME
    record_transforms_non_stream:
      - assign_roles

---

Run it

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

You should see users, departments, and roles load, followed by the hooks running. Any user ID in the assignments CSV that doesn’t match a user in users.csv will log a warning — this is expected behavior.

---

What’s next

This completes the CSV track. The next track moves to live REST APIs: authenticating with OAuth2 client credentials, paginating through API responses, and writing a custom connection module.

Next: Part 5 — Connecting to a REST API

See also

• Exhaust config.yaml before writing a hook

• How to use record transform hooks

• How to use pre-transform hooks

• How to use hooks to filter data