How to Enable Common Hooks

TL;DR

“Common hooks” are framework-owned features you turn on by name in config.yaml — the same way you enable a transform — without writing any integration code. The first built-in common hook, concurrent_fetch, fetches a source’s pages in parallel across a thread pool.

What is a common hook?

A common hook is a reusable behaviour that ships with oaa_generic and is enabled by name. Both the bare-string and name/options forms are valid:

    common_hooks:
      - concurrent_fetch                 # shorthand, default options
      - name: concurrent_fetch           # with options
        options:
          max_workers: 20
          page_size: 1000

Where you can declare them

Common hooks can be declared at three config levels, which set what they are scoped to:

# GLOBAL — applies across the whole run (every source)
common_hooks:
  - name: filter_stream_by_field
    options: {field: active, value: "true", action: keep}

sources:
  users:
    source_type: custom
    connection: api
    # SOURCE — scoped to just this source
    common_hooks:
      - name: filter_stream_by_field
        options: {field: kind, value: user, action: keep}

connections:
  api:
    connection_type: common
    # CONNECTION — applies to every source that uses this connection
    common_hooks:
      - name: concurrent_fetch
        options: {max_workers: 20}

There are three kinds of common hook:

• Fetch-phase hooks (like concurrent_fetch) replace how a source’s data is fetched. They are consulted automatically when the runner fetches each source (a source-level declaration overrides the connection’s).

• Lifecycle hooks bind to a lifecycle event (PRE_TRANSFORM … POST_PUSH_OAA) the hook itself declares, and run at that point scoped to their declaration level (global → all sources, source → that source, connection → sources using it).

• Push-phase hooks (like sqlite_streaming_push) replace how a datasource’s payload is serialized and pushed to Veza. Declared globally; the runner delegates the push to it instead of the default in-memory path.

Memory-bounded push for large populations

sqlite_streaming_push avoids the out-of-memory peak that large runs hit during JSON payload creation. The default push calls get_payload() (building the entire payload dict in memory) then json.dumps() (the entire string) on top of the live object graph. This hook instead serializes the provider into a SQLite database one object at a time, optionally frees the in-memory objects, then streams the payload from SQLite straight into a multipart upload — so the full dict and JSON string are never materialized.

common_hooks:
  - name: sqlite_streaming_push
    options:
      free_memory: true     # clear in-memory objects after staging
      part_size: 50000000   # multipart chunk size in bytes

It logs process RSS at each phase (before staging, after staging, after freeing, before/after push) so you can see where memory peaks. v1 supports CustomApplication providers. See the built-ins table in Common hooks for all options.

The concurrent_fetch built-in

Sequential pagination fetches page 0, then page 1, then page 2 … For large tables this is slow. concurrent_fetch fetches page 0 to learn the total record count, then dispatches all remaining pages across max_workers threads, turning an O(pages) wall-clock fetch into roughly O(pages / max_workers).

Options

Option	Default	Description
max_workers	1	Thread-pool size. 1 runs sequentially.
page_size	1000	Records requested per page.
limit_param	limit	Query param naming the page size.
offset_param	offset	Query param naming the row offset.
count_header	X-Total-Count	Response header carrying the total count.
count_path	none	JSON key carrying the total count (used if no header).
results_path	none	JSON key holding the records list (e.g. result). Omit if the body is itself the list.
base_url	connection uri	Base URL; ${ENV_VAR} references are interpolated.
path / endpoint	none	Path appended to base_url. {name} placeholders are filled from the source’s params (so one hook serves many sources).
params	none	Static query params added to every request (${ENV_VAR} interpolated).
headers	none	Static headers added to each request (${ENV_VAR} interpolated).
auth	none	Declarative auth block — see below.

Spilling very large sources to disk

A source can set spill_to_disk: true to stream its fetched rows into a temporary SQLite file instead of caching them in RAM — concurrent_fetch honours the flag automatically. Use it for very large destination: none junction sources (e.g. multi-million-row membership or role-assignment tables) that are consumed only by hooks and never become OAA objects; holding those in memory is a common cause of out-of-memory runs.

sources:
  memberships:
    destination: none
    source_type: custom
    connection: api
    spill_to_disk: true        # rows go to disk, not RAM
    params:
      resource: sys_user_grmember

The source is still fetched once and the table stays re-iterable (hooks that read it more than once just re-query SQLite), so it is a drop-in change. See spill_dict_table() for the underlying primitive and how to use it from a custom fetch().

Not sure which sources are worth spilling? Profile the run first — see How to Profile Memory Usage — and spill the sources whose transform:<source> step shows the biggest jump.

Environment variables and secrets

Reference secrets with ${ENV_VAR} in the options above and declare them with required_env on the connection so the run fails fast if any are missing. ${VAR} substitution is a config-wide feature (it works in every field, not just hook options) — see Environment variable substitution in the configuration reference for the full behavior.

Authentication

The auth block supports three declarative types, each with ${ENV_VAR} interpolation on its values:

# Static / env bearer token
auth:
  type: bearer
  token: "${MY_API_TOKEN}"

# HTTP basic
auth:
  type: basic
  username: "${MY_USER}"
  password: "${MY_PASS}"

# OAuth2 client credentials (token cached + refreshed on 401)
auth:
  type: oauth_client_credentials
  token_url: "https://${INSTANCE}/oauth_token.do"
  client_id: "${CLIENT_ID}"
  client_secret: "${CLIENT_SECRET}"
  scope: "optional scope"

Example: a ServiceNow-style table API, fully in config

This replicates the bespoke DCPS multithreaded fetch module — OAuth, instance, and credentials all sourced from the environment — with no integration code:

Use connection_type: common for a connection whose fetching is provided entirely by a fetch-phase common hook — it needs no fetch_module or custom_module:

connections:
  api:
    connection_type: common
    required_env:
      - DCPS_INSTANCE
      - DCPS_CLIENT_ID
      - DCPS_CLIENT_SECRET
    common_hooks:
      - name: concurrent_fetch
        options:
          base_url: "https://${DCPS_INSTANCE}"
          path: "/api/now/table/{resource}"   # resource comes from each source
          max_workers: 20
          page_size: 1000
          limit_param: sysparm_limit
          offset_param: sysparm_offset
          count_header: X-Total-Count
          results_path: result
          params:
            sysparm_display_value: "false"
          auth:
            type: oauth_client_credentials
            token_url: "https://${DCPS_INSTANCE}/oauth_token.do"
            client_id: "${DCPS_CLIENT_ID}"
            client_secret: "${DCPS_CLIENT_SECRET}"

sources:
  users:
    destination: LocalUser
    source_type: api
    connection: api
    params:
      resource: sys_user            # fills {resource} in the path

When to use the helper instead

The concurrent_fetch common hook covers offset pagination with bearer/basic/ OAuth-client-credentials auth. If your integration has a different auth flow (e.g. token exchange, mTLS) or non-offset pagination (cursor/page-number), call the underlying helper directly from your own fetch():

from oaa.modules.concurrent_fetch import concurrent_paginate
from oaa.modules.lazy_dicts import lazy_dict_table


def fetch(connection=None, source=None, **kwargs):
    api = MyAuthenticatedAPI(connection, source)
    table_name = source.params.get('resource')

    # Probe page 0 to learn the total, then fetch the rest concurrently.
    first_page, total = api.first_page(table_name, page_size=1000)

    def fetch_page(offset, page_size):
        return api.get_page(table_name, offset=offset, page_size=page_size)

    return lazy_dict_table(concurrent_paginate(
        fetch_page,
        total=total,
        page_size=1000,
        max_workers=20,
        first_page=first_page,
    ))

concurrent_paginate takes a fetch_page(offset, page_size) callable, so your auth and session handling stay entirely under your control. Pass a probed first_page so page 0 is not fetched twice.

Other built-in common hooks

Beyond concurrent_fetch, the lifecycle hook filter_stream_by_field keeps or drops source rows by a field value (a PRE_TRANSFORM hook, typically declared per source). See the full built-ins table and options in Common hooks.

Resolving secrets from an external credential store

The credential_store lifecycle hook retrieves a secret from an external credential store and writes it into an environment variable before the run starts (it binds to PRE_RUN). The resolved value is then available to ${ENV_VAR} interpolation in connection auth blocks, required_env, and get_oaa_client() (which reads os.environ['VEZA_API_KEY'] live at call time).

Skip-when-set behaviour

If the target env_var is already set in the environment (or .env file), the backend is not called at all — the run proceeds with the existing value. This keeps local and mocked runs from making unnecessary network calls to the credential store.

Config example (BeyondTrust Password Safe)

common_hooks:
  - name: credential_store
    options:
      backend: beyondtrust
      env_var: VEZA_API_KEY
      base_url: "${BT_BASE_URL}"
      api_user: "${BT_API_USER}"
      api_key: "${BT_API_KEY}"
      system_id: "${BT_SYSTEM_ID}"
      account_id: "${BT_ACCOUNT_ID}"

All option values support ${ENV_VAR} interpolation, so the credential store’s own credentials can be sourced from a .env file or shell environment.

Options

Option	Required	Description
backend	Yes	Name of the registered backend (e.g. beyondtrust).
env_var	Yes	Environment variable to write the retrieved secret into.
base_url	Yes (BeyondTrust)	Base URL of the credential store API.
api_user	Yes (BeyondTrust)	API user for authentication.
api_key	Yes (BeyondTrust)	API key for authentication.
system_id	Yes (BeyondTrust)	BeyondTrust system identifier.
account_id	Yes (BeyondTrust)	BeyondTrust account identifier.

The retrieved secret is never logged or printed — only written into os.environ.

Adding a custom backend

New credential-store backends can be registered without modifying the hook’s dispatch logic. Use the @credential_store_backend decorator:

from oaa.hooks.common.credential_store import credential_store_backend

@credential_store_backend('my_store')
def my_backend(options):
    # options is a dict of backend-specific keys (already interpolated)
    return fetch_secret(options)

Reference the backend by name in config: backend: my_store.

Related guides

• Configuration Schema Reference — Common hooks

• How to Handle API Pagination

• How to Write a Fetch Function

• How to Add a New Transform