Building Your First OAA Integration¶

A complete walkthrough of building an OAA integration from scratch. This tutorial uses BongaWonga as an example. It is a fictional SaaS platform that exposes user, role, and group data via REST API.

By the end of this tutorial, you’ll have a working integration that:

Fetches users and their attributes from an API
Extracts roles from user data
Fetches groups and enriches them with member information
Links users to roles and groups
Pushes everything to Veza in OAA format

TL;DR¶

This is a comprehensive tutorial. Expect 30-60 minutes to complete. By the end, you’ll understand the full lifecycle of building an OAA integration.

Prerequisites¶

Before starting, you’ll need:

Python 3.9+ installed
uv package manager installed (see How to Install uv)
Basic understanding of REST APIs and HTTP
Access to BongaWonga API documentation (provided in this tutorial)
A code editor (VS Code, PyCharm, etc.)

What is an OAA Integration?¶

An OAA (Open Authorization API) integration extracts identity and access data from a system (like BongaWonga) and transforms it into a standardized format that Veza can understand and process.

The transformation pipeline:

BongaWonga API → Fetch Data → Transform Fields → Build Relationships → Push to Veza

Part 1: Project Setup¶

Let’s start by creating the project structure.

Create Directory Structure¶

mkdir -p bongawonga/modules
cd bongawonga

Your directory should look like this:

bongawonga/
├── config.yaml           # Main configuration file
├── .env                  # Credentials (never commit)
├── .gitignore            # Exclude .env and __pycache__
└── modules/
    ├── app_bw.py         # API client
    └── app_bw_hooks.py   # Transformation hooks

Create .gitignore¶

Create a .gitignore file to prevent committing secrets:

cat > .gitignore << 'EOF'
# Environment variables and credentials
.env
.env.local

# Python
__pycache__/
*.pyc
*.pyo
*.egg-info/
.Python
venv/

# IDE
.vscode/
.idea/
*.swp

# Logs
logs/
*.log

# Output
output.json
EOF

Create Environment Variables File¶

Create a .env file with placeholder values (you’ll fill these in later):

cat > .env << 'EOF'
# BongaWonga App API Configuration
BW_APP_CLIENT_ID=your_auth0_client_id_here
BW_APP_CLIENT_SECRET=your_auth0_client_secret_here

# Veza Configuration (optional for local testing)
VEZA_URL=https://your-veza-instance.vezacloud.com
VEZA_API_KEY=your_veza_api_key_here
EOF

Important: Never commit .env to version control!

Create Minimal config.yaml¶

Create a basic configuration file:

cat > config.yaml << 'EOF'
version: v2
app:
  name: BongaWonga App
  application_type: Custom App
module_directory: ./modules
logic_modules:
  - app_bw_hooks
EOF

This minimal config:

Uses oaa-runner v2 format
Names the app “BongaWonga App”
Points to our modules directory
Imports our hooks module (which we’ll create next)

Now let’s create the API client!

Part 2: Define Your First Source - Users¶

Now we’ll add the users source to config.yaml. Update your config file:

version: v2
app:
  name: BongaWonga App
  application_type: Custom App
module_directory: ./modules
logic_modules:
  - app_bw_hooks


# Data sources
sources:
  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
      first_name:
        destination_field: first_name
        custom_type: STRING
      last_name:
        destination_field: last_name
        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

# Connection to BongaWonga API
connections:
  bw_app:
    connection_type: custom
    custom_module: app_bw
    required_env:
      - BW_APP_CLIENT_ID
      - BW_APP_CLIENT_SECRET
    params:
      base_url: https://app.bongawonga.com/api/v1
      dataset_id: "1"
      auth0_token_url: https://bongawonga.us.auth0.com/oauth/token
      auth0_audience: bongawonga-enterprises

This config:

Defines a custom connection to BongaWonga using Auth0 OAuth2
Adds a users source that maps to LocalUser (user identities in Veza)
Maps API fields to OAA fields (BongaWonga uses snake_case field names)
Includes a date conversion transform and lowercase normalization for email

Part 3: Create the API Client Module¶

The API client handles all communication with BongaWonga. We’ll build it incrementally.

Understanding the BongaWonga API¶

BongaWonga provides endpoints for users, roles, and groups:

Endpoint	Purpose	Authentication
`POST /oauth/token` (Auth0)	Get access token	Client ID + Secret + Audience
`GET /users?page=1&per_page=100`	Fetch paginated users	Bearer token
`GET /roles?page=1&per_page=100`	Fetch paginated roles	Bearer token
`GET /groups?page=1&per_page=100`	Fetch paginated groups	Bearer token
`GET /groups/{id}/members`	Fetch group members	Bearer token

Authentication: Auth0 OAuth 2.0 Client Credentials flow

POST https://bongawonga.us.auth0.com/oauth/token
{
  "grant_type": "client_credentials",
  "client_id": "...",
  "client_secret": "...",
  "audience": "bongawonga-enterprises"
}

Response:
{
  "access_token": "eyJhbGc...",
  "token_type": "Bearer",
  "expires_in": 3600
}

Pagination: Page + Per Page with has_next flag

GET /users?page=1&per_page=100
GET /users?page=2&per_page=100  # Next page
# Pagination response includes: {"pagination": {"has_next": true/false}}

Create modules/app_bw.py¶

This module handles all API interaction. Create the file:

cat > modules/app_bw.py << 'EOF'
"""BongaWonga API Client

Handles Auth0 authentication, pagination, and data fetching from BongaWonga API.
"""

import logging
import time

import requests

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:
    """Client for BongaWonga API"""

    def __init__(self, connection, source):
        """
        Initialize the API client.

        Args:
            connection: Connection config from config.yaml
            source: Source config with parameters
        """
        self.connection = connection
        self.source = source
        self.base_url = params_or_env(connection, source, 'base_url')
        self.dataset_id = params_or_env(connection, source, 'dataset_id', default='1')
        self.client_id = params_or_env(connection, source, 'BW_APP_CLIENT_ID')
        self.client_secret = params_or_env(connection, source, 'BW_APP_CLIENT_SECRET')
        self.token_url = params_or_env(connection, source, 'auth0_token_url')
        self.audience = params_or_env(connection, source, 'auth0_audience', default='')
        self._session = None
        self._access_token = None
        self._token_expires_at = 0.0

    def _get_access_token(self):
        """Get Auth0 OAuth2 access token using client credentials."""
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret,
        }
        if self.audience:
            payload['audience'] = self.audience

        logger.info('Requesting access token from %s', self.token_url)
        response = requests.post(self.token_url, data=payload, timeout=30)
        response.raise_for_status()
        data = response.json()
        self._access_token = data['access_token']
        expires_in = data.get('expires_in', 3600)
        self._token_expires_at = time.time() + expires_in - 60
        logger.info('Obtained access token (expires in %ss)', expires_in)

    @property
    def session(self):
        """Lazy initialization of HTTP session with auth headers"""
        if not self._access_token or time.time() >= self._token_expires_at:
            self._get_access_token()
        if not self._session:
            api_url = f"{self.base_url}/{self.dataset_id}"
            self._session = BaseURLSession(base_url=api_url)
        self._session.headers.update({
            'Authorization': f'Bearer {self._access_token}',
        })
        return self._session

    def fetch_paginated(self, path, resource_key='data', per_page=100):
        """
        Fetch paginated data from API.

        Generator yields records one at a time for memory efficiency.

        Args:
            path (str): API endpoint path (e.g., '/users')
            resource_key (str): JSON key containing records (default: 'data')
            per_page (int): Records per page (default: 100)

        Yields:
            dict: Individual record from API
        """
        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

            if response.status_code == 401:
                logger.warning('Received 401, re-authenticating')
                self._access_token = None
                self._get_access_token()
                self._session.headers.update({
                    'Authorization': f'Bearer {self._access_token}',
                })
                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):
    """
    Main entry point called by oaa-runner.

    Args:
        connection: Connection config from config.yaml
        source: Source config
        **kwargs: Additional arguments

    Returns:
        PETL table: Table of records
    """
    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)
EOF

This module:

Implements Auth0 OAuth2 client credentials authentication
Provides fetch_paginated() for memory-efficient data fetching with page-based pagination
Handles rate limiting (429 responses) and re-authentication (401 responses)
Exports a fetch() function that oaa-runner calls
Uses lazy_dict_table for memory efficiency

Test Running Your Integration¶

Now test your integration:

uv tool run --from oaa-runner \
  oaa app run --config-file=config.yaml --save-json

What this does:

Loads your config.yaml
Imports your modules
Connects to BongaWonga API via Auth0
Fetches users
Applies field mappings
Generates output.json with OAA-formatted data

Expected output in logs:

INFO: Requesting access token from https://bongawonga.us.auth0.com/oauth/token
INFO: Obtained access token (expires in 3600s)
DEBUG: Fetching from resource: /users
INFO: Success

Check the generated output.json:

# View the structure
jq '.' output.json | head -50

# Count users
jq '.local_users | length' output.json

# View first user
jq '.local_users[0]' output.json

You should see user records with the fields you mapped!

Part 4: Field Mapping¶

Field mapping is where source data becomes OAA-compatible data. Let’s deepen our mappings.

OAA Required Fields¶

For LocalUser, Veza requires:

unique_id (string) - Unique identifier for the user (required, must be unique)
name (string) - Display name (optional but highly recommended)
email (string) - Email address (optional)

Everything else is optional and can be custom fields.

Update Field Mappings¶

Let’s expand our users source with more sophisticated mappings:

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: bw_app
    params:
      resource: /users
      resource_key: data
    field_mapping:
      # Required fields
      unique_id:
        destination_field: unique_id
      email:
        destination_field: email
        transforms:
          - name: lowercase
      
      # Standard fields
      first_name:
        destination_field: first_name
        custom_type: STRING
      last_name:
        destination_field: last_name
        custom_type: STRING
      
      # Combine first + last name into display name
      name:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.first_name }} {{ row.last_name }}"
      
      # Boolean fields - convert status to is_active
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
      
      # Date fields
      created_at:
        destination_field: created_at
        transforms:
          - name: date_convert
      hire_date:
        destination_field: hire_date
        transforms:
          - name: date_convert
      
      # Custom fields
      department:
        destination_field: department
        custom_type: STRING
      title:
        destination_field: title
        custom_type: STRING
      manager_id:
        destination_field: manager_id
        custom_type: STRING
      phone:
        destination_field: phone
        custom_type: STRING

Understanding the Transforms¶

jinja_render: Creates a new field using a Jinja2 template. Here we combine first and last names.
validate_boolean: Converts various truthy/falsy values (yes/no, true/false, 1/0, active/inactive) to actual booleans.
date_convert: Parses dates and converts to ISO 8601 format compatible with Veza.

Test the Updated Mapping¶

Run the integration again:

uv tool run --from oaa-runner \
  oaa app run --config-file=config.yaml --save-json

Verify the new fields:

# View first user with all fields
jq '.local_users[0]' output.json

# Check that name field was created properly
jq '.local_users[] | {id: .unique_id, name: .name}' output.json | head -20

Excellent! Now let’s add roles.

Part 5: Add a Derived Source - Roles¶

Roles are often embedded in user data. Rather than making a separate API call, we can “derive” roles from the users source. This is more efficient.

Add Roles Source to config.yaml¶

Add this after your users source definition:

  roles:
    destination: LocalRole
    source_type: source
    source: users
    field_mapping:
      name:
        destination_field: name
      unique_id:
        destination_field: unique_id
    record_transforms_non_stream:
      - map_roles_to_users

This config:

Uses source_type: source to indicate it derives from another source
References source: users - uses users as input
Maps role fields to OAA fields (using BongaWonga snake_case naming)
Declares a record transform hook to link roles to users

Create Transformation Hooks¶

Now create modules/app_bw_hooks.py:

cat > modules/app_bw_hooks.py << 'EOF'
"""BongaWonga Integration Hooks

Hooks for data transformation, enrichment, and relationship building.
"""

import logging
from oaa.hooks.decorators import hook, OAAHookEvent, record_transform_non_stream

logger = logging.getLogger(__name__)


@hook(event=OAAHookEvent.PRE_TRANSFORM, target='roles')
def extract_roles_from_users(event, source=None, provider=None, **kwargs):
    """
    Extract role records from user data.

    Users may have roles embedded in their data. This hook flattens them into individual role records.
    """
    
    def extract_role(row):
        """Extract a role record from a user row"""
        if hasattr(row, 'role_name') and row.role_name:
            yield (row.role_name, row.role_name)
    
    source._stream = source.stream.rowmapmany(
        extract_role,
        header=['name', 'unique_id']
    )
    
    logger.debug(f"Extracted roles from {source.source} users")


@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM, target='roles')
def map_roles_to_users(record=None, provider=None, **kwargs):
    """Link roles to users via the Provider object"""
    try:
        user = provider.local_users.get(record.unique_id)
        
        if not user:
            logger.warning(f"User {record.unique_id} not found for role {record.name}")
            return
        
        user.add_role(record.name, apply_to_application=True)
        logger.debug(f"Added role {record.name} to user {record.unique_id}")
    
    except Exception as e:
        logger.error(f"Error mapping role {record.name}: {e}")
EOF

This file contains:

extract_roles_from_users(): A PRE_TRANSFORM hook that converts user records into role records using BongaWonga field names
map_roles_to_users(): A RECORD_TRANSFORM hook that links users to their roles

Test with Roles¶

Run the integration:

uv tool run --from oaa-runner \
  oaa app run --config-file=config.yaml --save-json

Inspect the roles and relationships:

# Count roles
jq '.local_roles | length' output.json

# View a role
jq '.local_roles[0]' output.json

# Check user-role relationships
jq '.relationships[] | select(.src_type=="LocalUser" and .dst_type=="LocalRole") | length' output.json

Great! Now let’s add teams/groups.

Part 6: Add Groups Source¶

Groups (teams) are fetched from a separate API endpoint and represent collections of users.

Add Groups Source to config.yaml¶

Add this after your roles source:

  groups:
    destination: LocalGroup
    source_type: custom
    connection: bw_app
    params:
      resource: /groups
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      name:
        destination_field: name
      description:
        destination_field: description
        custom_type: STRING
      owner_id:
        destination_field: owner_id
        custom_type: STRING
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
    record_transforms_non_stream:
      - map_group_members

This defines:

A custom source that fetches from /groups (BongaWonga naming)
Field mappings to convert group data to LocalGroup
A record transform to link users to groups

Add Group Membership Hook¶

Add this function to modules/app_bw_hooks.py:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='groups')
def fetch_group_members(event, source=None, **kwargs):
    """
    Enrich group data with membership information.
    
    For each group, fetch the list of member user IDs from the API.
    """
    from modules.app_bw import BongaWongaAPI
    from oaa.settings import config
    
    api = BongaWongaAPI(
        config.connections[source.connection],
        source
    )
    
    def get_members(row):
        """Fetch members for a group"""
        try:
            path = f"/groups/{row.unique_id}/members"
            response = api.session.get(path, timeout=30)
            data = response.json()
            
            members = data.get('data', [])
            member_ids = [m['unique_id'] for m in members]
            
            logger.debug(f"Group {row.unique_id} has {len(member_ids)} members")
            
            return member_ids
        except Exception as e:
            logger.warning(f"Could not fetch members for group {row.unique_id}: {e}")
            return []
    
    source._stream = source.stream.addfield('members', get_members)
    logger.debug("Enriched groups with membership data")


@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM, target='groups')
def map_group_members(record=None, provider=None, **kwargs):
    """Link users to groups via the Provider object"""
    try:
        group = provider.local_groups.get(record.unique_id)
        
        if not group:
            logger.warning(f"Group {record.unique_id} not found in provider")
            return
        
        # Add group owner
        if hasattr(record, 'owner_id') and record.owner_id:
            owner = provider.local_users.get(record.owner_id)
            if owner:
                owner.add_group(record.unique_id)
                logger.debug(f"Added owner {record.owner_id} to group {record.unique_id}")
        
        # Add all group members
        members = getattr(record, 'members', [])
        for member_id in members:
            user = provider.local_users.get(member_id)
            if user:
                user.add_group(record.unique_id)
        
        logger.debug(f"Added {len(members)} members to group {record.unique_id}")
    
    except Exception as e:
        logger.error(f"Error mapping group members for {record.unique_id}: {e}")

Add these functions to your app_bw_hooks.py file.

Test with Groups¶

Run the integration:

uv tool run --from oaa-runner \
  oaa app run --config-file=config.yaml --save-json

Verify groups and group memberships:

# Count groups
jq '.local_groups | length' output.json

# View a group
jq '.local_groups[0]' output.json

# Check user-group relationships
jq '.relationships[] | select(.src_type=="LocalUser" and .dst_type=="LocalGroup")' output.json | head -20

Perfect! Now let’s test everything end-to-end with better debugging.

Part 7: Run and Debug¶

Let’s run the complete integration with debugging tools.

Run the Complete Integration¶

uv tool run --from oaa-runner \
  oaa app run \
  --config-file=config.yaml \
  --save-json \
  --debug

Flags explained:

--config-file=config.yaml: Your configuration file
--save-json: Saves output to a local JSON file (useful for inspection)
--debug: Enable debug logging for detailed output

Inspect the Output¶

Use jq to query the output:

# Summary of all entities
jq '{
  users: (.local_users | length),
  roles: (.local_roles | length),
  groups: (.local_groups | length),
  relationships: (.relationships | length)
}' output.json

# View all users (paginated)
jq '.local_users | length' output.json
jq '.local_users[] | {id: .unique_id, name: .name, email: .email}' output.json | head -20

# View all roles
jq '.local_roles | length' output.json
jq '.local_roles[]' output.json

# View all groups
jq '.local_groups | length' output.json
jq '.local_groups[]' output.json

# Count relationships by type
jq '.relationships | group_by(.src_type) | map({type: .[0].src_type, count: length})' output.json

# View relationships for a specific user
jq '.relationships[] | select(.src_id=="user123")' output.json

Use Priority Push for Faster Testing¶

When testing against a real Veza instance, use --priority-push to get faster feedback:

uv tool run --from oaa-runner \
  oaa app run \
  --config-file=config.yaml \
  --save-json \
  --priority-push

This prioritizes your push in Veza’s processing queue, so you see results faster during development.

Common Errors and Fixes¶

Error	Cause	Fix
`ModuleNotFoundError: No module named 'app_bw'`	Module not in `logic_modules`	Add to logic_modules list in config.yaml
`KeyError: 'users'`	Field mapping references non-existent field	Check API response field names with jq
`AttributeError: 'NoneType' object has no attribute 'unique_id'`	Mapping missing required unique_id field	Ensure unique_id is mapped in field_mapping
`Connection refused`	Can’t reach API endpoint	Check base_url in config.yaml connections block
`401 Unauthorized`	Bad credentials	Verify BW_APP_CLIENT_ID and BW_APP_CLIENT_SECRET in .env
`PETL error: field 'X' not found`	Field missing from source data	Check source API actually returns this field

Part 8: Production Readiness¶

Before deploying to production, make these improvements.

Add Error Handling¶

Update your API client to handle errors gracefully:

# In modules/app_bw.py, enhance the session property with error handling:

@property
def session(self):
    """Lazy initialization of HTTP session with auth headers and error handling"""
    if not self._access_token or time.time() >= self._token_expires_at:
        self._get_access_token()
    if not self._session:
        api_url = f"{self.base_url}/{self.dataset_id}"
        self._session = BaseURLSession(base_url=api_url)
    self._session.headers.update({
        'Authorization': f'Bearer {self._access_token}',
    })
    return self._session

Handle Rate Limits¶

The API client already handles 429 (too many requests) with exponential backoff. For production, you might want to make this more sophisticated:

def get(self, path, params=None, retries=3):
    """GET with configurable retry logic"""
    for attempt in range(retries):
        try:
            response = self.session.get(path, params=params, timeout=30)
            
            if response.status_code == 429:
                wait_time = (2 ** attempt) * 5  # Exponential backoff
                logger.warning(f"Rate limited, waiting {wait_time}s (attempt {attempt+1}/{retries})")
                time.sleep(wait_time)
                continue
            
            response.raise_for_status()
            return response
        
        except Exception as e:
            if attempt == retries - 1:
                raise
            logger.warning(f"Attempt {attempt+1} failed: {e}")
            time.sleep(2 ** attempt)
    
    raise Exception(f"Failed to fetch {path} after {retries} attempts")

Use lazy_dict_table for Memory Efficiency¶

Your code already uses lazy_dict_table, which is excellent. This ensures large datasets don’t cause memory issues.

Add Structured Logging¶

Use logging effectively to track what’s happening:

import logging

# Configure logging with timestamps and levels
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

# Use appropriate log levels:
logger.debug("Detailed diagnostic information")
logger.info("General informational messages")
logger.warning("Warning messages (unexpected but not errors)")
logger.error("Error messages")

Validate Required Fields¶

Add validation to ensure critical fields are present:

def validate_record(record, required_fields):
    """Validate that required fields are present"""
    for field in required_fields:
        if not getattr(record, field, None):
            logger.error(f"Missing required field '{field}' in record: {record}")
            return False
    return True

# In your PRE_TRANSFORM hooks:
@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def validate_users(event, source=None, **kwargs):
    required = ['unique_id', 'email']
    
    def filter_valid(row):
        return validate_record(row, required)
    
    source._stream = source.stream.select(filter_valid)

Complete Code Reference¶

Here’s the complete code for reference.

config.yaml (Complete)¶

version: v2
app:
  name: BongaWonga App
  application_type: Custom App
module_directory: ./modules
logic_modules:
  - app_bw_hooks

connections:
  bw_app:
    connection_type: custom
    custom_module: app_bw
    required_env:
      - BW_APP_CLIENT_ID
      - BW_APP_CLIENT_SECRET
    params:
      base_url: https://app.bongawonga.com/api/v1
      dataset_id: "1"
      auth0_token_url: https://bongawonga.us.auth0.com/oauth/token
      auth0_audience: bongawonga-enterprises

sources:
  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
      first_name:
        destination_field: first_name
        custom_type: STRING
      last_name:
        destination_field: last_name
        custom_type: STRING
      name:
        destination_field: name
        transforms:
          - name: jinja_render
            options:
              template: "{{ row.first_name }} {{ row.last_name }}"
      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
      hire_date:
        destination_field: hire_date
        transforms:
          - name: date_convert
      department:
        destination_field: department
        custom_type: STRING
      title:
        destination_field: title
        custom_type: STRING
      manager_id:
        destination_field: manager_id
        custom_type: STRING
      phone:
        destination_field: phone
        custom_type: STRING

  roles:
    destination: LocalRole
    source_type: source
    source: users
    field_mapping:
      name:
        destination_field: name
      unique_id:
        destination_field: unique_id
    record_transforms_non_stream:
      - map_roles_to_users

  groups:
    destination: LocalGroup
    source_type: custom
    connection: bw_app
    params:
      resource: /groups
      resource_key: data
    field_mapping:
      unique_id:
        destination_field: unique_id
      name:
        destination_field: name
      description:
        destination_field: description
        custom_type: STRING
      owner_id:
        destination_field: owner_id
        custom_type: STRING
      status:
        destination_field: is_active
        transforms:
          - name: jinja_render
            options:
              template: "{{ 'true' if row.status == 'active' else 'false' }}"
          - name: validate_boolean
    record_transforms_non_stream:
      - map_group_members

modules/app_bw.py (Complete)¶

See Part 3 above for the complete implementation.

modules/app_bw_hooks.py (Complete)¶

"""BongaWonga Integration Hooks

Hooks for data transformation, enrichment, and relationship building.
"""

import logging
from oaa.hooks.decorators import hook, OAAHookEvent, record_transform_non_stream

logger = logging.getLogger(__name__)


@hook(event=OAAHookEvent.PRE_TRANSFORM, target='roles')
def extract_roles_from_users(event, source=None, provider=None, **kwargs):
    """
    Extract role records from user data.

    Users may have roles. This hook flattens them into individual role records.
    """
    
    def extract_role(row):
        """Extract a role record from a user row"""
        if hasattr(row, 'role_name') and row.role_name:
            yield (row.role_name, row.role_name)
    
    source._stream = source.stream.rowmapmany(
        extract_role,
        header=['name', 'unique_id']
    )
    
    logger.debug(f"Extracted roles from users")


@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM, target='roles')
def map_roles_to_users(record=None, provider=None, **kwargs):
    """Link roles to users via the Provider object"""
    try:
        user = provider.local_users.get(record.unique_id)
        
        if not user:
            logger.warning(f"User {record.unique_id} not found for role {record.name}")
            return
        
        user.add_role(record.name, apply_to_application=True)
        logger.debug(f"Added role {record.name} to user {record.unique_id}")
    
    except Exception as e:
        logger.error(f"Error mapping role {record.name}: {e}")


@hook(event=OAAHookEvent.PRE_TRANSFORM, target='groups')
def fetch_group_members(event, source=None, **kwargs):
    """Enrich group data with membership information"""
    from modules.app_bw import BongaWongaAPI
    from oaa.settings import config
    
    api = BongaWongaAPI(
        config.connections[source.connection],
        source
    )
    
    def get_members(row):
        """Fetch members for a group"""
        try:
            path = f"/groups/{row.unique_id}/members"
            response = api.session.get(path, timeout=30)
            data = response.json()
            
            members = data.get('data', [])
            member_ids = [m['unique_id'] for m in members]
            
            logger.debug(f"Group {row.unique_id} has {len(member_ids)} members")
            
            return member_ids
        except Exception as e:
            logger.warning(f"Could not fetch members for group {row.unique_id}: {e}")
            return []
    
    source._stream = source.stream.addfield('members', get_members)
    logger.debug("Enriched groups with membership data")


@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM, target='groups')
def map_group_members(record=None, provider=None, **kwargs):
    """Link users to groups via the Provider object"""
    try:
        group = provider.local_groups.get(record.unique_id)
        
        if not group:
            logger.warning(f"Group {record.unique_id} not found in provider")
            return
        
        # Add group owner
        if hasattr(record, 'owner_id') and record.owner_id:
            owner = provider.local_users.get(record.owner_id)
            if owner:
                owner.add_group(record.unique_id)
                logger.debug(f"Added owner {record.owner_id} to group {record.unique_id}")
        
        # Add group members
        members = getattr(record, 'members', [])
        for member_id in members:
            user = provider.local_users.get(member_id)
            if user:
                user.add_group(record.unique_id)
        
        logger.debug(f"Added {len(members)} members to group {record.unique_id}")
    
    except Exception as e:
        logger.error(f"Error mapping group members for {record.unique_id}: {e}")

Next Steps¶

Congratulations! You’ve built a complete OAA integration. Here’s what to do next:

Explore Advanced Patterns¶

Chained Sources — Derive data from other sources without extra API calls
Async Enrichment — Fetch additional data in parallel using async workers
Mock Modules — Create mock data for testing without API credentials
Custom Transforms — Build Python functions for complex transformations

Deploy to Production¶

Set up proper error handling and logging
Configure rate limit backoff
Use environment variables for credentials
Test with real data from your system
Set up monitoring and alerting

Appendix: Common Patterns¶

Pattern: Handling Multiple Roles Per User¶

If users can have multiple roles, use rowmapmany:

def extract_roles(row):
    """Yield one role record per role in user's role list"""
    for role in row.roles:  # Assuming roles is a list
        yield (role['name'], row.userId)

source._stream = source.stream.rowmapmany(
    extract_roles,
    header=['roleName', 'userId']
)

Pattern: Conditional Field Mapping¶

Use Jinja transforms for conditional logic:

displayName:
  destination_field: name
  transforms:
    - name: jinja_render
      options:
        template: |
          {% if row.userFirstName and row.userLastName %}
            {{ row.userFirstName }} {{ row.userLastName }}
          {% elif row.userFirstName %}
            {{ row.userFirstName }}
          {% else %}
            {{ row.userEmail }}
          {% endif %}

Pattern: Rate-Limited Parallel Requests¶

For fetching multiple resources per record, use asyncio:

import asyncio
from aiohttp import ClientSession

async def fetch_all_async(items, fetch_func):
    """Fetch multiple items in parallel with concurrency limit"""
    async with ClientSession() as session:
        tasks = [fetch_func(session, item) for item in items]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

This completes the tutorial! You now understand the full lifecycle of building an OAA integration.