How to Use RECORD_TRANSFORM Hooks

TL;DR

Transform individual records after field mapping has been applied — build group memberships, role assignments, and other relationships that config.yaml field mapping can’t express, or validate/enrich records with external lookups.

Adding a custom property? Use custom_type in config.yaml field_mapping instead of a hook. See Exhaust config.yaml before writing a hook.

Prerequisites

  • An OAA integration with config.yaml and sources

  • Field mapping already configured

  • A modules/hooks.py file for hook implementations

  • Understanding that RECORD_TRANSFORM runs AFTER field mapping (unlike PRE_TRANSFORM)

Steps

Step 1: Create RECORD_TRANSFORM Hook

In modules/hooks.py, add a RECORD_TRANSFORM hook:

"""Integration hooks for record-level transformations"""
import logging
from oaa.hooks.decorators import record_transform_non_stream, OAAHookEvent

logger = logging.getLogger(__name__)


@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def transform_user_record(record=None, provider=None, **kwargs):
    """
    Transform individual user records after field mapping.
    
    Called for EACH record after field mapping is applied.
    This runs on LocalUser objects that have been created.
    
    Args:
        record: A single user record with OAA fields already mapped
        provider: The provider object (for accessing other records)
        **kwargs: Additional arguments
    """
    
    # Validate required fields
    if not record.email:
        logger.warning(f"User {record.unique_id} has no email")
        record.email = f"noemail_{record.unique_id}@company.com"
    
    # Clean up data
    if hasattr(record, 'name') and record.name:
        record.name = record.name.strip()
    
    return record

Step 2: Register Hook in config.yaml

Ensure your hooks module is registered:

module_directory: ./modules
logic_modules:
  - hooks  # Your hooks module
  - my_api_client

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /users
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email
      name:
        destination_field: name
       department:
         destination_field: department
         custom_type: STRING  # Declared here, not in a hook — this is how department becomes a custom property

For all available custom_type options, see What are Veza Custom Properties?.

Step 3: Validate and Test

Test your record transformation:

# Validate configuration
uv tool run --from oaa-runner oaa app validate --config=config.yaml

# Run integration
uv tool run --from oaa-runner oaa app run --config=config.yaml --save-json

# Check that custom properties were added
jq '.local_users[0] | .custom_properties' output.json

Common Record Transformations

Data Validation

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def validate_records(record=None, **kwargs):
    """Validate and fix common data issues"""
    
    # Validate email format
    if record.email and '@' not in record.email:
        logger.error(f"Invalid email for {record.unique_id}: {record.email}")
        record.email = ""
    
    # Ensure name is not empty
    if not record.name or record.name.strip() == "":
        record.name = record.unique_id
    
    # Convert to string if needed
    if hasattr(record, 'phone_number') and record.phone_number:
        record.phone_number = str(record.phone_number)
    
    return record

Anti-pattern: Don’t Add Custom Properties in RECORD_TRANSFORM

It’s tempting to reach for a hook whenever a source field needs a bit of massaging, but simple field-derived values belong in config.yaml, not in a hook:

# ❌ Don't do this — RECORD_TRANSFORM is the wrong place for plain custom properties
@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def add_custom_properties(record=None, **kwargs):
    if hasattr(record, 'department') and record.department:
        record.add_custom_property('department', record.department,
                                  property_type='string')
    return record

Instead, declare department (and any other field) directly in field_mapping with custom_type, as shown in Step 2 above. See Exhaust config.yaml before writing a hook and What are Veza Custom Properties?.

RECORD_TRANSFORM should be reserved for things field_mapping genuinely can’t do — chiefly building relationships (group memberships, role assignments, manager/sub-resource edges) that require looking up other records via provider.

Enrich with a Lookup and Wire a Relationship

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def assign_role_from_lookup(record=None, **kwargs):
    """Resolve a role name from an ID and assign it to the record"""
    
    # Example: lookup role from ID
    if hasattr(record, 'role_id'):
        role_mapping = {
            '1': 'Admin',
            '2': 'Manager',
            '3': 'User'
        }
        
        role_name = role_mapping.get(str(record.role_id), 'Unknown')
        record.add_role(role_name)
    
    return record

Clean and Normalize Data

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def clean_data(record=None, **kwargs):
    """Normalize and clean data"""
    
    # Trim whitespace from text fields
    if record.name:
        record.name = record.name.strip()
    
    if record.email:
        record.email = record.email.strip().lower()
    
    # Convert empty strings to None
    if hasattr(record, 'phone_number') and record.phone_number == '':
        record.phone_number = None
    
    # Normalize boolean values
    if hasattr(record, 'is_active'):
        if isinstance(record.is_active, str):
            record.is_active = record.is_active.lower() in ('true', 'yes', '1')
    
    return record

Examples

Real Integration: Genetec — Group Membership

From qts/genetec/modules/genetec_hooks.py:

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def user_to_group_mapping(record=None, provider=None, **kwargs):
    """Add each user to their Genetec group via the provider lookup"""
    
    group_guid = record.get('group_guid')
    if not group_guid:
        return record
    
    user = provider.local_users.get(record.get('user_guid'))
    if user:
        user.add_group(group_guid)
    
    return record

Real Integration: Wolken — Role Assignment

From cloud_software_group/wolken/modules/wolken_hooks.py:

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def map_roles_to_users(record=None, provider=None, **kwargs):
    """Assign the resolved role to the user record"""
    
    user = provider.local_users.get(record.get('user_id'))
    if user and record.get('role_name'):
        user.add_role(record.get('role_name'))
    
    return record

Using the Provider Object

Access other records through the provider:

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def add_manager_relationships(record=None, provider=None, **kwargs):
    """Add manager relationships using provider"""
    
    if not provider:
        return record
    
    # If this record has a manager ID, link to manager
    if hasattr(record, 'manager_id') and record.manager_id:
        try:
            # Get manager from provider and add a relationship
            manager = provider.local_users.get(record.manager_id)
            if manager:
                # For custom app users, model reporting via roles/groups or custom properties
                logger.debug(f"Manager {record.manager_id} found for {record.unique_id}")
        except KeyError:
            logger.warning(f"Manager {record.manager_id} not found")
    
    return record

Hook Event Timing

RECORD_TRANSFORM runs after field mapping:

1. FETCH: Get raw data
    ↓
2. PRE_TRANSFORM: Modify stream (optional)
    ↓
3. FIELD_MAPPING: Auto-map fields
    ↓
4. Create OAA objects
    ↓
5. RECORD_TRANSFORM ← You are here! (with provider access)
    ↓
6. PRE_PUSH_OAA: Build relationships
    ↓
7. Push to Veza

Use RECORD_TRANSFORM when you need to:

  • ✅ Access mapped OAA fields

  • ✅ Validate individual records

  • ✅ Build group/role/relationship edges via provider

  • ✅ Access provider object to resolve other records

  • ✅ Make decisions based on transformed data

Don’t use RECORD_TRANSFORM to add plain custom properties — declare those in config.yaml field_mapping instead.

Return Values

Your RECORD_TRANSFORM hook must return the record:

@record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)
def transform_record(record=None, **kwargs):
    # Modify the record
    record.name = record.name.strip()
    
    # MUST return the record
    return record

Troubleshooting

Error: AttributeError: 'LocalUser' object has no attribute 'X'

  • Verify the field is in your field_mapping

  • Check that source provides this field

  • Use hasattr(record, 'X') to check before accessing

Relationships not appearing

  • Verify the provider lookup (provider.local_users.get(...), etc.) returns the expected object, not None

  • Check that the target group/role/resource exists before referencing it

  • Verify the hook is registered in logic_modules

Custom property not appearing

Hook not being called

  • Check decorator: @record_transform_non_stream(event=OAAHookEvent.RECORD_TRANSFORM)

  • Verify module is in logic_modules list

  • Check logs for import errors

Next Steps