What are OAA Runner Hooks?

Use OAA Runner hooks to insert custom logic into various stages of the data stream. This is commonly used to modify data prior to being sent to Veza.

@hook()

@record_transform(stream=True|False)

Available Hook Events

Hook Event Summary

Event	When it fires
PRE_FETCH	Before any connection.fetch() call for a source. Use to inject auth headers, modify params, etc. Scoped via target_connection= kwarg.
PRE_TRANSFORM	Before field mapping/transforms run on a source. Use to reshape or filter the raw data stream.
POST_TRANSFORM	After transforms are queued but before data is read into memory. Field mappings have already been applied.
PRE_CREATE_OAA	After transforms. Use to build relationships between objects (e.g. user → role) before the push.
PRE_PUSH_OAA	Just before the OAA payload is sent to Veza. All data is in memory at this point.
POST_PUSH_OAA	After the payload has been accepted by Veza. The hook receives the Veza API response.
POST_DB_CONNECT	Fires after a successful database connection is established. Receives the live connection object.
RECORD_TRANSFORM_STREAM	Applied per-record on the petl stream. Register via @record_transform(stream=True) (the default).
RECORD_TRANSFORM	Applied per-record after OAA objects are created. Register via @record_transform(stream=False).

IDE Linter Caution

Warning

IDE linters (Pylance, Pyright, mypy) may auto-correct hook function signatures, changing:

• source=None → table=None

• provider=None → app=None

• removing guard clauses (if source is None)

These auto-corrections break hooks silently at runtime.

Workaround: Write hook files directly via Bash heredoc or disable linting on hook files:

cat > modules/myapp_hooks.py << 'EOF'
from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_TRANSFORM, target="users")
def filter_users(source=None, **kwargs):  # ← Linter won't modify this
    ...
EOF

Or add a linter directive at the top of the file:

# type: ignore
# pylint: disable=all

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def wire_roles(provider=None, sources=None, **kwargs):
    ...

Always verify hook signatures in your final code before testing.

Hook Parameter Reference

Each hook event receives different parameters. Use only the relevant ones:

Critical Rules

Warning

• target= kwarg is ONLY valid on PRE_TRANSFORM, POST_TRANSFORM, PRE_CREATE_OAA, RECORD_TRANSFORM. It has NO effect on PRE_PUSH_OAA or POST_PUSH_OAA.

• target_connection= kwarg is ONLY valid on PRE_FETCH. It scopes the hook to sources using a specific connection name. Both target= and target_connection= can be combined on PRE_FETCH.

• Always use provider=None in PRE_PUSH_OAA/POST_PUSH_OAA signatures (NOT app=None). Using app=None causes all provider calls to silently fail.

• RECORD_TRANSFORM does NOT receive connection kwarg. Access connection params via config.connections["conn_name"] instead.

• PRE_TRANSFORM targeting a derived source (source_type: source) must use source=None parameter (NOT table=None).

Example — Correct PRE_PUSH_OAA Signature:

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def wire_roles(provider=None, sources=None, config=None, **kwargs):
    """Wire users to their roles."""
    for user in provider.local_users.values():
        user.add_role("admin_role_id")

Example — Correct RECORD_TRANSFORM Signature:

@hook(event=OAAHookEvent.RECORD_TRANSFORM)
def enrich_user(record=None, source=None, config=None, **kwargs):
    """Enrich user record with custom properties."""
    record.custom_department = record.get("department", "Unknown")
    return record

Detailed Examples Per Hook Event

This section provides complete examples for each major hook event, showing parameter access and common patterns.

PRE_TRANSFORM — Filter or Reshape Raw Stream

Runs before field mapping transforms. Use to filter records or reshape the raw stream before transformations are applied.

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_TRANSFORM, target="users")
def filter_active_users(source=None, **kwargs):
    """Filter stream to active users only."""
    # source is the Source object for "users"
    # source.stream is the petl stream (not yet field-mapped)

    def is_active(record):
        # Access raw API field names here, not destination names
        return record.get("status") == "active"

    # Replace stream with filtered version
    import petl
    source._stream = petl.select(source.stream, is_active)

PRE_FETCH — Inject Auth or Modify Connection Before Fetch

Runs before any connection.fetch() call, for all source types (CSV, API, Veza API, custom modules). Use to inject authentication headers, modify connection parameters, or log fetch events. Scope to a specific connection via target_connection=.

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_FETCH, target_connection="my_api")
def inject_bearer_token(connection=None, source=None, **kwargs):
    """Inject a bearer token before the API call."""
    connection.auth = {
        "token": os.environ.get("MY_API_TOKEN"),
    }

@hook(event=OAAHookEvent.PRE_FETCH)
def log_all_fetches(connection=None, source=None, config=None, **kwargs):
    """Log every fetch regardless of connection."""
    logger.info("Fetching source %s via connection %s",
                source.source_name,
                getattr(connection, "name", None))

When to Choose PRE_FETCH vs Fetch-Phase Common Hooks

Use PRE_FETCH when you need to augment the existing fetch — inject an auth header, tweak connection params, or log the request. No YAML changes required; everything lives in Python.

Use common hooks (@common_hook(phase='fetch') + YAML config) when you need to replace the fetch entirely — for example, using concurrent_fetch for parallel pagination, or a custom fetch implementation that doesn’t fit the standard API fetch pattern. Common hooks completely bypass connection.fetch().

In short:

• Augment the fetch → PRE_FETCH

• Replace the fetch → common_hook(phase='fetch')

POST_TRANSFORM — Access Transformed Data

Runs after field mapping transforms are queued but before data is fully read into memory. Access transformed field names.

@hook(event=OAAHookEvent.POST_TRANSFORM, target="users")
def validate_transformed_data(source=None, **kwargs):
    """Validate field mappings produced expected output."""
    # At this point, field mappings have been applied to headers
    # but data hasn't been fully consumed yet

    # Get the transformed field names
    header = source.stream.header()

    # Check that required destination fields exist
    assert "unique_id" in header, "unique_id not in field mapping"
    assert "email" in header, "email not in field mapping"

PRE_CREATE_OAA — Access All Loaded Data Before OAA Objects

Runs after all sources are loaded and transformed, before OAA objects are created. Can access multiple sources.

@hook(event=OAAHookEvent.PRE_CREATE_OAA)
def validate_data_integrity(sources=None, **kwargs):
    """Validate data across multiple sources before OAA creation."""
    # sources is dict[str, Source] — access by name

    users_source = sources["users"]
    roles_source = sources["roles"]

    # Count records in each source
    user_count = len(users_source.records)
    role_count = len(roles_source.records)

    print(f"Loading {user_count} users and {role_count} roles")

    # Check for consistency: every user role must exist in roles source
    user_roles = set()
    for user in users_source.records:
        user_roles.add(user.get("role_id"))

    existing_roles = set()
    for role in roles_source.records:
        existing_roles.add(role.get("unique_id"))

    missing_roles = user_roles - existing_roles
    if missing_roles:
        print(f"Warning: {len(missing_roles)} user roles not found: {missing_roles}")

RECORD_TRANSFORM — Per-Record Processing with Provider Access

Runs after OAA objects are created. Can access provider object to look up already-created entities.

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.RECORD_TRANSFORM)
def enrich_and_link(record=None, source=None, config=None, provider=None, **kwargs):
    """Add custom properties and link to other objects."""
    # record is the OAA object (e.g., LocalUser)
    # source is the Source this record came from
    # config gives access to all connection params
    # provider has all loaded OAA objects

    # Add custom property
    record.department = source.records.get("department", "Unknown")

    # Look up related object
    manager_id = record.get("manager_id")
    if manager_id:
        manager = provider.local_users.get(manager_id)
        if manager:
            record.manager_email = manager.email

    return record

PRE_PUSH_OAA — Wire Relationships and Final Transformations

Runs after all objects are created and transformed, just before pushing to Veza. Use to build relationships and make final adjustments.

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_PUSH_OAA)
def wire_relationships(provider=None, config=None, **kwargs):
    """Wire users to roles and groups; final permission checks."""
    # provider has all OAA objects: local_users, local_roles, local_groups
    # sources is NOT available in PRE_PUSH_OAA

    # Wire users to their roles
    for user in provider.local_users.values():
        # Look up the user's assigned role
        role_name = user.properties.get("assigned_role")

        # Find the corresponding role object
        matching_role = None
        for role in provider.local_roles.values():
            if role.name == role_name:
                matching_role = role
                break

        if matching_role:
            user.add_role(matching_role.unique_id, apply_to_application=True)

    # Wire users to their groups
    for user in provider.local_users.values():
        department = user.properties.get("department")

        # Find groups by name
        for group in provider.local_groups.values():
            if group.name == department:
                user.add_group(group.unique_id)

    # Validate before push
    user_count = len(provider.local_users)
    role_count = len(provider.local_roles)
    group_count = len(provider.local_groups)

    print(f"PRE_PUSH_OAA: {user_count} users, {role_count} roles, {group_count} groups")

POST_PUSH_OAA — Cleanup and Logging

Runs after Veza accepts the payload. Use for cleanup, logging, or external notifications.

@hook(event=OAAHookEvent.POST_PUSH_OAA)
def log_push_results(provider=None, **kwargs):
    """Log successful push and cleanup."""
    # provider still has all OAA objects (but they're now submitted)
    # kwargs may include 'veza_response' with push status

    user_count = len(provider.local_users)
    role_count = len(provider.local_roles)

    print(f"Successfully pushed {user_count} users, {role_count} roles to Veza")

    # Example: cleanup temporary files
    import os
    if os.path.exists("temp_data.json"):
        os.remove("temp_data.json")