How to Use PRE_TRANSFORM Hooks

TL;DR

Modify or enrich a data stream before field mapping is applied, such as adding calculated fields, flattening nested data, or filtering records.

Prerequisites

  • An OAA integration set up with config.yaml

  • A source fetching data (REST API, database, CSV, etc.)

  • Basic understanding of PETL operations

  • A modules/hooks.py file for hook implementations

Steps

Step 1: Create hooks.py

Create modules/hooks.py in your integration:

"""Integration hooks for data transformation"""
import logging
from oaa.hooks.decorators import hook, OAAHookEvent

logger = logging.getLogger(__name__)


@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def transform_users(event, source=None, **kwargs):
    """
    Transform raw user data before field mapping.
    
    Args:
        event: Hook event information
        source: The data source being transformed
        **kwargs: Additional arguments
    """
    # Modify the stream by adding fields, renaming, etc.
    source._stream = source.stream \
        .addfield('full_name', lambda v: f"{v.first_name} {v.last_name}") \
        .cutout('internal_id')  # Remove fields not needed
    
    logger.info("✓ Transformed user data stream")

Step 2: Register Hooks in config.yaml

Add your hooks module to the logic_modules list:

version: v2
app:
  name: MyIntegration
  application_type: MyIntegration
module_directory: ./modules
logic_modules:
  - hooks  # Register 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
      full_name:  # This field is created by the hook!
        destination_field: name

Step 3: Use PETL Operations to Transform

Common PETL operations for data transformation:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def transform_users(event, source=None, **kwargs):
    """Example of various PETL transformations"""
    
    source._stream = source.stream \
        # Add calculated fields
        .addfield('full_name', lambda v: f"{v.first_name} {v.last_name}") \
        .addfield('domain', lambda v: v.email.split('@')[1]) \
        .addfield('is_active', lambda v: v.status.lower() == 'active') \
        \
        # Rename fields
        .rename('user_id', 'uid') \
        .rename('email_address', 'email') \
        \
        # Remove unwanted fields
        .cutout('internal_id', 'temp_field', 'debug_info') \
        \
        # Convert data types
        .convert('created_date', lambda v: str(v) if v else '') \
        \
        # Filter rows
        .select(lambda v: v.status != 'deleted')
    
    logger.info("✓ Applied PETL transformations")

Step 4: Test Your Hooks

Validate that your hooks work:

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

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

# Inspect output.json to verify transformations
jq '.local_users[0]' output.json

Common Transformations

Flatten Nested Data

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def flatten_nested_fields(event, source=None, **kwargs):
    """Extract values from nested dictionaries"""
    
    source._stream = source.stream \
        .addfield('city', lambda row: row.location.get('city', '')
                  if isinstance(row.location, dict) else '') \
        .addfield('country', lambda row: row.location.get('country', '')
                  if isinstance(row.location, dict) else '') \
        .cutout('location')

Add Derived Fields

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def add_derived_fields(event, source=None, **kwargs):
    """Add calculated/derived fields"""
    
    source._stream = source.stream \
        # Combine fields
        .addfield('full_name', 
                 lambda v: f"{v.first_name} {v.last_name}") \
        \
        # Extract domain from email
        .addfield('email_domain', 
                 lambda v: v.email.split('@')[1] if '@' in v.email else '') \
        \
        # Boolean flags
        .addfield('is_admin', 
                 lambda v: v.role.lower() == 'admin') \
        \
        # Date formatting
        .addfield('created_year',
                 lambda v: v.created_date[:4] if v.created_date else '')

Filter and Clean Data

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def filter_and_clean(event, source=None, **kwargs):
    """Remove invalid records and clean data"""
    
    source._stream = source.stream \
        # Remove deleted users
        .select(lambda v: v.status != 'deleted') \
        \
        # Remove records with missing email
        .select(lambda v: v.email and '@' in v.email) \
        \
        # Trim whitespace
        .addfield('email', lambda v: v.email.strip()) \
        .addfield('name', lambda v: v.name.strip() if v.name else '') \
        \
        # Log filtered count
        
    logger.info("✓ Filtered and cleaned data stream")

Production Examples

Pattern 1: Flatten Nested Data (Abnormal Security)

From gxo/abnormal/modules/abnormal_hooks.py - transforms nested API response data:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def transform_users(event, source=None, **kwargs):
    """Transform Abnormal nested user data into flat structure"""
    
    def flatten_user(row):
        # Extract email domain for filtering/analysis
        if hasattr(row, 'email'):
            parts = row.email.split('@')
            row.email_local = parts[0]
            row.email_domain = parts[1] if len(parts) > 1 else ''
            # Add risk calculation
            row.risk_level = 'high' if row.email_domain in ['suspicious-domain.com'] else 'low'
        return row
    
    source._stream = source.stream \
        .addfield('flattened', flatten_user) \
        .cutout('flattened')  # Remove temporary field
    
    logger.info("✓ Transformed Abnormal user data")

Use case: When your API returns nested objects or complex structures that need to be flattened for the OAA data model.

Pattern 2: Multi-Source Data Combination (IHG Centrify)

From ihg/centrify/modules/centrify_v5.py - combines data from multiple API calls:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def enhance_user_data(event, source=None, **kwargs):
    """Combine user data with metadata from separate endpoints"""
    
    # Cache data from parallel sources
    user_metadata = fetch_user_metadata()  # Additional API call
    
    def enrich_user(row):
        # Add metadata if available
        if row.user_id in user_metadata:
            meta = user_metadata[row.user_id]
            row.department = meta.get('department', '')
            row.office_location = meta.get('location', '')
            row.last_login = meta.get('last_login', '')
        return row
    
    source._stream = source.stream \
        .addfield('enriched', enrich_user) \
        .cutout('enriched')
    
    logger.info("✓ Enriched user data from metadata")

Use case: When you need to combine data from multiple sources or API endpoints to create richer user profiles.

Pattern 3: Filter and Transform (ThousandEyes)

From netapp/thousandeyes/modules/thousandeyes_hooks.py - filters and cleans data:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')
def transform_users(event, source=None, **kwargs):
    """Clean and filter ThousandEyes user data"""
    
    source._stream = source.stream \
        # Only keep active users
        .select(lambda v: v.status == 'active') \
        # Add display name from email if name not available
        .addfield('display_name',
                 lambda v: v.name or v.email.split('@')[0]) \
        # Remove internal/sensitive fields
        .cutout('internal_id', 'api_token', 'secret', 'password_hash')
    
    logger.info("✓ Cleaned ThousandEyes user data")

Use case: When you need to filter out inactive, deleted, or test records and remove sensitive fields.

Hook Event Timing

PRE_TRANSFORM runs at the right time:

1. FETCH: Get raw data from source
    ↓
2. PRE_TRANSFORM ← You are here!
    ↓
3. FIELD_MAPPING: Map to OAA fields
    ↓
4. Create OAA objects
    ↓
5. RECORD_TRANSFORM: Per-record logic (with provider access)
    ↓
6. PRE_PUSH_OAA: Build relationships
    ↓
7. Push to Veza

Use PRE_TRANSFORM when you need to:

  • ✅ Modify the entire data stream

  • ✅ Add or remove fields

  • ✅ Flatten nested data

  • ✅ Filter records

  • ✅ Prepare data before mapping

Troubleshooting

Error: AttributeError: 'stream' object has no attribute 'addfield'

  • Verify you’re using PETL operations correctly

  • Check that source._stream = ... is being called

  • Import PETL if using advanced operations

Error: Hook not running

  • Verify module name is in logic_modules list in config.yaml

  • Check hook decorator: @hook(event=OAAHookEvent.PRE_TRANSFORM, target='users')

  • Verify target matches source name exactly

Transformed data not appearing in output

  • Check logs for errors during transformation

  • Verify field is referenced in field_mapping

  • Make sure field name is correct after transformation

Next Steps