Learning the OAA Runner Framework: Part 10 — Linking Identities Across Systems¶

Three identical Calvins standing in a row, each holding a different ID badge ("APP ID", "HRIS ID", "IDP ID"). They are connected together by strings tied around their wrists. Hobbes stands to the side looking very confused, pointing at all three of them.

This is Part 10 of the Learning the OAA Runner Framework series. It builds on Part 6 — Multiple Systems, Multiple Destinations and Part 9 — Defending Against Bad Data.

What you’ll learn: Use a shared identifier to join user records from two systems — the BongaWonga App and HRIS APIs — and enrich app users with HRIS data. You’ll load HRIS data inside a hook using a module-level cache, so the HRIS API is called once per run rather than once per user.

Prerequisites¶

Completed or read Part 6 — Multiple Systems, Multiple Destinations and Part 9 — Defending Against Bad Data.
API credentials for the BongaWonga App and HRIS APIs. Get them from each system’s profile page — see Getting API credentials.

Setup¶

Create your working directory:

mkdir -p 10-identity-linking/modules
cd 10-identity-linking

Set your API keys:

export BW_APP_API_KEY=your_app_api_key
export BW_HRIS_API_KEY=your_hris_api_key

Get each key from the corresponding system’s profile page:

Copy the following code into a new file at modules/app_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 BongaWongaAPI:
    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_APP_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 = BongaWongaAPI(connection, source)
    resource_path = source.params.get('resource', '')
    resource_key = source.params.get('resource_key', 'data')
    logger.debug('Fetching from 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/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/link_hooks.py:

import logging
import os
from types import SimpleNamespace

from oaa.hooks.decorators import record_transform_non_stream
from hris_bw import HrisAPI

logger = logging.getLogger(__name__)

_hris_by_employee_number = {}


def _load_hris_lookup():
    """Load all HRIS employees into a dict keyed by employee_number.

    Uses a module-level cache so the API is called only once across all
    invocations of the hook for different app user records.
    """
    if _hris_by_employee_number:
        return _hris_by_employee_number
    conn = SimpleNamespace(params={
        'base_url': os.environ.get('BW_HRIS_BASE_URL', 'https://hris.bongawonga.com/api/v1'),
    })
    api = HrisAPI(conn, source=None)
    for emp in api.fetch_paginated('/employees'):
        num = emp.get('employee_number')
        if num:
            _hris_by_employee_number[num] = emp
    logger.info('Loaded %d HRIS employees into lookup', len(_hris_by_employee_number))
    return _hris_by_employee_number


@record_transform_non_stream(target='app_users')
def enrich_with_hris(record, provider=None, **kwargs):
    """Enrich app users with HRIS data using external_user_id as the join key."""
    lookup = _load_hris_lookup()
    ext_id = getattr(record, 'external_user_id', None)
    if not ext_id:
        logger.debug(
            'User %s has no external_user_id — skipping HRIS enrichment', record.unique_id
        )
        return
    emp = lookup.get(ext_id)
    if not emp:
        logger.debug('No HRIS match for external_user_id %s', ext_id)
        return
    user = provider.local_users.get(record.unique_id)
    if not user:
        return
    if emp.get('department'):
        user.department = emp['department']
    if emp.get('hire_date'):
        user.hire_date = emp['hire_date']
    logger.debug(
        'Enriched user %s with HRIS data (employee_number=%s)', record.unique_id, ext_id
    )

Create an empty config.yaml:

touch config.yaml

Step 1: The cross-system key map¶

The three BongaWonga systems share user identity through a combination of fields:

System	Identifier field	Shared with
App	`external_user_id`	matches HRIS `employee_number`
HRIS	`employee_number`	matches App `external_user_id`
IdP	`external_id`	also matches `employee_number`
All three	`email`	common across all systems

In this example, the join key between App and HRIS is: app.external_user_id == hris.employee_number

Email is a fallback link — Veza can use it to correlate identities across providers. The external_user_id field is the more reliable programmatic key here because it’s set explicitly by the provisioning system.

Step 2: Add a second connection to the config¶

In Parts 5–6, the config had one connection. This config has two — one for the App API and one for the HRIS API:

connections:
  bw_app:
    connection_type: custom
    custom_module: app_bw
    required_env:
      - BW_APP_API_KEY
    params:
      base_url: https://app.bongawonga.com/api/v1

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

Important: In this example, bw_hris is defined as a connection but is not used as the source for any OAA entities in the config. HRIS data does not load as Employee records. Instead, the link_hooks.py module fetches HRIS data directly inside the hook using the HrisAPI class from hris_bw.py. The bw_hris connection in config is present solely for required_env validation. oaa-runner checks required_env across all connection definitions at startup — before any source loads or hook runs. Declaring BW_HRIS_API_KEY here ensures a clear error at launch if that variable is missing, rather than a runtime crash inside the hook when the HRIS fetch is attempted.

Step 3: The module-level cache pattern¶

Open modules/link_hooks.py. The hook runs once per app user record. If it called the HRIS API once per user, a dataset with 10,000 users would make 10,000 API requests — slow and likely to hit rate limits.

The solution is a module-level dict that acts as a cache:

_hris_by_employee_number = {}

def _load_hris_lookup():
    if _hris_by_employee_number:
        return _hris_by_employee_number
    # ... fetch all employees and populate the dict ...
    return _hris_by_employee_number

Because the dict lives at module level (not inside the function or class), it persists for the lifetime of the Python process. The first invocation fetches all HRIS employees and builds the dict. Every subsequent invocation returns the already-populated dict immediately.

Why not load it at module import time? Loading at import time means the HRIS API would be called when the module is imported, before oaa-runner has validated credentials or set up logging. Loading lazily on first hook invocation keeps the fetch inside the normal run lifecycle.

Note: The module-level dict persists for the lifetime of the Python process — not just one oaa-runner invocation. If oaa-runner is called multiple times in the same process (for example, from a scheduler or wrapper script), the cache from the first run is reused on subsequent runs. HRIS data will be stale for those later runs. If data freshness matters, add a TTL check or clear the dict at the start of each run.

Step 4: Review link_hooks.py¶

modules/link_hooks.py was created in Setup. Here is the full file:

import logging
import os
from types import SimpleNamespace
from oaa.hooks.decorators import record_transform_non_stream
from hris_bw import HrisAPI

logger = logging.getLogger(__name__)

_hris_by_employee_number = {}

def _load_hris_lookup():
    if _hris_by_employee_number:
        return _hris_by_employee_number
    conn = SimpleNamespace(params={
        'base_url': os.environ.get('BW_HRIS_BASE_URL', 'https://hris.bongawonga.com/api/v1'),
    })
    api = HrisAPI(conn, source=None)
    for emp in api.fetch_paginated('/employees'):
        num = emp.get('employee_number')
        if num:
            _hris_by_employee_number[num] = emp
    logger.info('Loaded %d HRIS employees into lookup', len(_hris_by_employee_number))
    return _hris_by_employee_number

@record_transform_non_stream(target='app_users')
def enrich_with_hris(record, provider=None, **kwargs):
    lookup = _load_hris_lookup()
    ext_id = getattr(record, 'external_user_id', None)
    if not ext_id:
        logger.debug(
            'User %s has no external_user_id — skipping HRIS enrichment', record.unique_id
        )
        return
    emp = lookup.get(ext_id)
    if not emp:
        logger.debug('No HRIS match for external_user_id %s', ext_id)
        return
    user = provider.local_users.get(record.unique_id)
    if not user:
        return
    if emp.get('department'):
        user.department = emp['department']
    if emp.get('hire_date'):
        user.hire_date = emp['hire_date']
    logger.debug(
        'Enriched user %s with HRIS data (employee_number=%s)', record.unique_id, ext_id
    )

Key points:

from hris_bw import HrisAPI — imports the class from the sibling module in ./modules. This works because module_directory is added to sys.path during hook execution. Two constraints apply: only top-level .py files in module_directory are importable (subdirectory packages are not on the path); and module filenames that collide with Python stdlib names or installed packages (e.g., os.py, requests.py) will shadow the standard module — name your files with a project-specific prefix to avoid this.
SimpleNamespace(params={...}) — constructs a lightweight connection object without an actual config.yaml connection definition. The HRIS API class expects connection.params. Why not use the bw_hris connection directly? Hooks are called with record and provider — they do not receive the config-defined connection objects. There is no way to pass connection into a hook. SimpleNamespace gives the HrisAPI constructor the object shape it expects (connection.params) built from env vars instead. The API key (BW_HRIS_API_KEY) does not need to appear in the SimpleNamespace params dict — params_or_env falls back to the environment automatically when a key is absent from connection.params.
The null guard on ext_id applies the pattern from Part 9: skip silently at DEBUG level when no join key is present.
os.environ.get('BW_HRIS_BASE_URL', 'https://hris.bongawonga.com/api/v1') — the base URL for the HRIS API. The code defaults to the standard BongaWonga HRIS URL if the env var is not set. You can override it by setting BW_HRIS_BASE_URL in your shell if your HRIS API is at a non-standard location. For most setups, the default is correct and you do not need to set this variable.
user.department = emp['department'] — sets the OAA field directly on the user object. Only overwrite if the HRIS value is non-empty, to avoid blanking out values already set by the app field mapping.

Step 5: Wire the hook into the config¶

Add logic_modules to the config header, expand app_users to carry all the standard user fields from Part 6, and add external_user_id as a new STRING custom field. Map record_transforms_non_stream on the source to run the hook after each user record loads:

module_directory: ./modules
logic_modules:
  - link_hooks

sources:
  app_users:
    destination: LocalUser
    source_type: custom
    connection: bw_app
    params:
      resource: /users
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      email:
        destination_field: email
        transforms:
          - name: lowercase
      name:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.display_name }}"
      first_name:
        destination_field: given_name
        custom_type: STRING
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.first_name }}"
      last_name:
        destination_field: family_name
        custom_type: STRING
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.last_name }}"
      title:
        destination_field: title
        custom_type: STRING
      department:
        destination_field: department
        custom_type: STRING
      user_type:
        destination_field: user_type
        custom_type: STRING
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
      created_at:
        destination_field: created_at
        transforms:
          - name: date_convert
      last_login:
        destination_field: last_login_at
        transforms:
          - name: date_convert
      external_user_id:
        destination_field: external_user_id
        custom_type: STRING
    record_transforms_non_stream:
      - enrich_with_hris

Complete config¶

version: v2
log_level: INFO

app:
  name: BongaWonga App (with HRIS enrichment)
  application_type: Custom App

module_directory: ./modules
logic_modules:
  - link_hooks

connections:
  bw_app:
    connection_type: custom
    custom_module: app_bw
    required_env:
      - BW_APP_API_KEY
    params:
      base_url: https://app.bongawonga.com/api/v1

  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:
  app_users:
    destination: LocalUser
    source_type: custom
    connection: bw_app
    params:
      resource: /users
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      email:
        destination_field: email
        transforms:
          - name: lowercase
      name:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.display_name }}"
      first_name:
        destination_field: given_name
        custom_type: STRING
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.first_name }}"
      last_name:
        destination_field: family_name
        custom_type: STRING
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.last_name }}"
      title:
        destination_field: title
        custom_type: STRING
      department:
        destination_field: department
        custom_type: STRING
      user_type:
        destination_field: user_type
        custom_type: STRING
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
      created_at:
        destination_field: created_at
        transforms:
          - name: date_convert
      last_login:
        destination_field: last_login_at
        transforms:
          - name: date_convert
      external_user_id:
        destination_field: external_user_id
        custom_type: STRING
    record_transforms_non_stream:
      - enrich_with_hris

Run it¶

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

On first run, the hook will fetch all HRIS employees and log:

INFO Loaded 8432 HRIS employees into lookup

Then for each app user with a matching employee_number, you’ll see:

DEBUG Enriched user u-1234 with HRIS data (employee_number=EMP-8821)

Users without a match log at DEBUG level and are skipped silently. Set log_level: DEBUG in the config to see the full trace.

Summary: cross-system identity key table¶

App field	HRIS field	Shared via
`external_user_id`	`employee_number`	Provisioning system sets the App field to match HRIS
`email`	`email`	Common across all three BongaWonga systems
(no direct field)	`external_id` on IdP	IdP also uses `employee_number` as its `external_id`

When building a cross-system integration:

Identify which field each system uses as its canonical user identifier.
Find the shared key that links them (often email, employee number, or SSO subject).
Map that key as a STRING custom property so hooks can read it.
Guard against null and empty-string cases before using it as a dict key.

What’s next¶

This is the final article in the BongaWonga example series. To go deeper:

Review the how-to guides for reference on individual features.
Explore the explanation articles for the reasoning behind OAA design decisions.