Configuration Schema Reference

The config.yaml file is the primary way to configure an OAA Runner integration. This reference documents every supported field, its type, whether it is required, and what it does.

For a conceptual overview see What is an OAA Config File?.

---

Top-level fields

Example:

version: v2
app:
  name: My Application
  application_type: custom
  url: https://myapp.example.com
  datasource: my-app-ds
sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /v1/users
    field_mapping:
      id:
        destination_field: unique_id
      email:
        destination_field: email
permission_to_veza_map:
  read:
    permissions:
      - DataRead
  write:
    permissions:
      - DataWrite
      - DataRead
run_params:
  dry_run: true        # parse data but don't push to Veza
  save_json: false     # save payload JSON to local file
  no_push: false       # skip all Veza API calls entirely
  priority_push: false # push with priority flag
connections:
  my_api:
    connection_type: custom
    custom_module: my_api
    required_env:
      - MY_API_KEY

Field	Type	Required	Default	Description
version	string	No	v2	Config format version. Always v2.
app	App	Yes*	—	Application definition. Required when runner_type=custom_app.
runner_type	string	No	custom_app	Runner type. Currently only custom_app is supported. Can be omitted.
VEZA_URL	string	No	env var	Base URL of your Veza instance (e.g. https://myorg.vezacloud.com). Can be set via environment variable.
VEZA_API_KEY	string	No	env var	Veza API key. Can be set via environment variable.
provider_icon_path	string	No	—	Path to a PNG or SVG icon (≤ 64 KB) used for the provider in Veza.
enable_multipart	bool	No	false	Enable multipart upload for large payloads.
run_params	RunParams	No	—	Runtime execution flags (dry_run, save_json, no_push, priority_push).
GROUP_TYPE	string	No	Department	Label used for employee group types (HRIS integrations).
destination_type	string	No	—	Provider template to use: CUSTOM_APP, IDP, HRIS, or ENTITY_ENRICHMENT. Controls which OAA object type is created. See How to enrich existing Veza entities for the enrichment flow.
idp_provider_type	string	No	okta	IdP provider type (okta, azure_ad, google_workspace, etc.).
module_directory	string	No	—	Path to a directory containing custom logic modules. Added to sys.path.
logic_modules	list[string]	No	[]	Python modules to load and run as post-processing logic hooks.
common_hooks	list[CommonHook]	No	[]	Global common hooks — framework-owned, reusable behaviors enabled by name, applied across the whole run. See Common hooks.
sources	dict[str, Source]	No	{}	Named data sources. Keys are source names used throughout the config.
connections	dict[str, Connection]	No	—	Named connections to data systems. Keys are referenced by source.connection.
permission_to_veza_map	dict[str, PermissionMapping]	No	{}	Maps application-specific permission names to Veza permission types.

---

app

Defines the application created in Veza.

Field	Type	Required	Description
name	string	Yes	Display name of the application in Veza.
application_type	string	No	Application type identifier.
domain	string	No	Domain (IdP integrations).
idp_type	string	No	IdP type (IdP integrations).
hris_type	string	No	HRIS type (HRIS integrations).
url	string	No	URL of the application.
options	dict	No	Application-specific options.
provisioning	bool	No	Whether the application supports provisioning.
external_lifecycle_management_type	string	No	—
data_plane_id	string	No	—
datasource	string	No	"{name} Datasource"

app:
  name: My Application
  application_type: custom
  url: https://myapp.example.com
  options:
    enable_sso: true
  provisioning: true
  external_lifecycle_management_type: SCIM
  datasource: my-app-ds

---

sources

Each key under sources is a named source. The value is a Source object.

Field	Type	Required	Description
source_type	string	Yes	One of CSV, CSV_PUSH, DB, API, VEZA_API, CUSTOM, SOURCE.
destination	string	No	Veza object type to create: LocalUser, LocalGroup, LocalRole, Resource, EnrichedEntity, none. Use EnrichedEntity only with destination_type: ENTITY_ENRICHMENT.
entity_type	string	Yes*	Required when destination: EnrichedEntity. Veza entity type to enrich (e.g. AwsIamRole, OktaUser).
data_source_id	string	Yes*	Required when destination: EnrichedEntity. ID of the Veza datasource that owns the entities being enriched (e.g. "339083562601:awsiam").
source_name	string	No	Human-readable name. Auto-set to the key name if omitted.
source	string	No	For source_type: SOURCE — references another source by name.
connection	string	No	Name of the connection to use (must match a key under connections).
csv_filepath	string	Yes*	Path to the CSV file. Required when source_type=CSV.
csv_provider_id	string	No	Veza provider ID. Required for source_type=CSV_PUSH.
mapping_filepath	string	Yes*	Path to a CSV field mapping file. Required when source_type is CSV, DB, API, or CUSTOM and field_mapping is not set.
field_mapping	dict[str, FieldMapping]	No	Inline field mappings (preferred over mapping_filepath).
resource_type	string	No	Resource type label for destination: Resource objects.
query	string	No	SQL query (used with source_type: DB).
tree	Tree	No	Hierarchical resource configuration.
params	dict	No	Arbitrary key/value parameters passed to the connection’s fetch().
record_transforms	list[string]	No	Transform function names applied per-record via the stream pipeline.
record_transforms_non_stream	list[string]	No	Transform function names applied per-record outside the stream pipeline.
stream_transforms	list[StreamTransform]	No	Declarative PETL transforms applied to the raw fetched stream, in order, before any PRE_TRANSFORM hooks run. See Reshaping Data with stream_transforms.
common_hooks	list[CommonHook]	No	Common hooks scoped to this source. See Common hooks.
spill_to_disk	bool	No	Stream this source’s fetched rows to a temporary SQLite file instead of caching them in RAM. Use for very large destination: none junction sources (e.g. multi-million-row membership/role tables) that are consumed only by hooks — keeps them off the heap (can save several GB at peak). Honoured by the concurrent_fetch hook; custom fetch() modules can call oaa.modules.spill_dict_table directly.
do_perm_mapping	bool	No	Auto-register custom permissions for destination: LocalRole sources. When true, each role’s name is used to register a custom permission (resolved from permission_to_veza_map) and attach it to the role. Falls back to NonData for unmapped roles.
attributes	dict[str, str]	No	Static attributes added to every record.
datasource	string	No	—

Examples by source type:

# CSV source
sources:
  users:
    destination: LocalUser
    source_type: csv
    csv_filepath: ./data/users.csv
    field_mapping:
      employee_id:
        destination_field: unique_id
      email:
        destination_field: email

# API source
sources:
  users:
    destination: LocalUser
    source_type: api
    connection: my_api
    params:
      resource: /v1/users
    field_mapping:
      id:
        destination_field: unique_id

# DB source
sources:
  users:
    destination: LocalUser
    source_type: db
    connection: my_db
    query: SELECT id, email, display_name FROM users

# CUSTOM source
sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_custom
    params:
      resource: /users

# SOURCE (derived) — filters/extends another source
sources:
  admins:
    source_type: source
    source: users
    attributes:
      role: admin

field_mapping entry (FieldMapping)

Each key in field_mapping is the source field name. The value can be:

1. A string — treated as the destination_field (shorthand)

2. A dict without destination_field — the key is used as destination_field (shorthand)

3. A full dict — explicit destination_field and other options

Field	Type	Required	Description
destination_field	string	Yes*	Veza field this source field maps to. *Required in full syntax; auto-inferred from key in shorthand forms.
source_field	string	No	Override the source field name (defaults to the map key).
map_to_identity	bool	No	When true, calls add_identity() with this field’s value on each LocalUser record. Empty and None values are skipped. Can be combined with destination_field and custom_type — identity assignment is additive. Only applies to LocalUser destinations. Default: false.
custom_type	string	No	Modern form. Creates a Veza custom property with this type. Values: STRING, NUMBER, BOOLEAN, TIMESTAMP. Preferred over is_custom/veza_field_type.
is_custom	bool	No	Legacy. Use custom_type instead. If true, creates a Veza custom property.
veza_field_type	string	No	Legacy. Use custom_type instead. Type of the custom property: string, number, boolean, timestamp.
transforms	list	No	List of transform functions to apply to the value.

Shorthand syntax (recommended for simple mappings):

field_mapping:
  # String shorthand — destination_field = "unique_id"
  id: unique_id

  # Dict shorthand — destination_field inferred as "department"
  department:
    source_field: dept
    custom_type: STRING

  # Empty dict — destination_field inferred as "email"
  email: {}

Full syntax (needed for transforms or explicit control):

field_mapping:
  user_id:
    destination_field: unique_id
  job_title:
    destination_field: title
    is_custom: true
    veza_field_type: string
  status:
    destination_field: account_status
    transforms:
      - name: lowercase

Both syntaxes can be mixed within the same field_mapping.

⚠️ Important: Duplicate Field Mapping Keys

YAML silently uses the LAST definition when the same source field key appears twice in field_mapping. This is a YAML parsing behavior, not a validation error, and results in silent mapping failures.

Incorrect (second firstName overwrites first):

field_mapping:
  firstName:
    destination_field: first_name
  firstName:  # ❌ YAML ignores the above entry
    destination_field: name

Correct (use aliases for multiple transforms on same source):

field_mapping:
  firstName_raw:
    source_field: firstName
    destination_field: first_name
  firstName_upper:
    source_field: firstName
    destination_field: name
    transforms:
      - name: uppercase

Or use Jinja2 templates in a single entry:

field_mapping:
  firstName:
    destination_field: first_name
  full_name:
    transforms:
      - name: jinja_render
        parameters:
          template: "{{row.firstName}} {{row.lastName}}"
    destination_field: name

Always verify that each unique source field appears exactly once as a key in your field_mapping.

Derived Sources (source_type: source)

When you need to create a new entity type from data already in another source (e.g., extract roles from users, derive departments from employee records), use source_type: source instead of making a second API call. This pattern reuses the parent source’s streaming data, avoiding redundant API requests and memory overhead.

Basic Pattern

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: myapp
    params:
      resource: /api/users
    field_mapping:
      id:
        destination_field: unique_id
      email:
        destination_field: email
      dept:
        destination_field: department
  
  # Derived source: extract roles from users
  roles:
    destination: LocalRole
    source_type: source              # ← Key: derives from users source
    source: users                     # ← Parent source name
    field_mapping:
      role_name:
        destination_field: name       # Map from user data
      role_id:
        destination_field: unique_id

Critical Rules

1. Parent source must be listed BEFORE dependent source

The source that is referenced (users) must appear before the source that derives from it (roles). OAA Runner loads sources in order, so dependents cannot reference a parent that hasn’t loaded yet.

2. Do NOT include connection: or params: on dependent source

Dependent sources do not fetch data. They receive the parent’s streaming PETL pipeline after field mappings have run. No credentials or API parameters are needed.

3. Dependent source receives parent’s transformed data

The dependent source gets the parent source’s data AFTER field mapping transforms have already been applied. Use the destination field names, not original source names:

sources:
  users:
    source_type: custom
    connection: myapp
    field_mapping:
      firstName:
        destination_field: first_name    # ← Transformed name
      lastName:
        destination_field: last_name      # ← Transformed name
  
  # Roles reads from transformed user data
  roles:
    source_type: source
    source: users
    field_mapping:
      first_name:                        # ← Uses destination_field name
        destination_field: created_by

Reshaping Data with PRE_TRANSFORM

If you need to extract, filter, or reorganize records from the parent stream, use a PRE_TRANSFORM hook targeting the dependent source:

from oaa.hooks import hook, OAAHookEvent

@hook(event=OAAHookEvent.PRE_TRANSFORM, target="roles")
def extract_roles_from_users(source=None, **kwargs):
    """Extract unique roles from user department field."""
    def gen_roles():
        seen = set()
        for record in source.stream.dicts():
            role_name = record.get("department")
            if role_name and role_name not in seen:
                yield {"role_name": role_name, "role_id": f"role_{role_name}"}
                seen.add(role_name)
    
    source._stream = source.stream.rowmapmany(
        lambda r: [{"role_name": r.get("department"), 
                   "role_id": f"role_{r.get('department')}"}],
        header=["role_name", "role_id"]
    )

Important details:

• Use @hook(event=OAAHookEvent.PRE_TRANSFORM, target="roles") — the target= kwarg scopes the hook to the “roles” source

• Accept source=None as a parameter (NOT table=None)

• Assign to source._stream — no return value

• Transform happens BEFORE field_mapping, so you work with the parent’s original field names, not destination names

Reshaping Data with stream_transforms

For reshaping that can be expressed as a fixed PETL pipeline, prefer the declarative stream_transforms list over a Python hook. Each entry is a single-key mapping: the key is the operation name and the value is the operation payload.

sources:
  users:
    destination: LocalRole
    source_type: custom
    connection: myapp
    stream_transforms:
      # Explode each user's groups into one row per group
      - rowmapmany:
          iterable:
            field: row.userGroups
          skip_if: item.id is none
          columns:
            role_id: item.id
            role_name: item.name
            role_owner: row.ownerUserId
      # De-duplicate on role_id (also sorts by role_id)
      - distinct: [role_id]
      # Keep only rows whose owner is set
      - select: row.role_owner
      # Reorder rows by role_name
      - sort: [role_name]
      # Drop fields that are not needed downstream
      - cutout: [raw_groups]

stream_transforms run on the raw fetched stream before any PRE_TRANSFORM hooks, so hooks always see the reshaped data.

Supported operations:

Op	Payload	Description
rowmap	RowmapConfig	One output row per input row. See below.
rowmapmany	RowmapConfig	Explode each input row into zero or more output rows (one per item in iterable).
distinct	field name / list	Remove duplicate rows on the given key. PETL distinct() sorts by the key, so output order is not the input order — chain a sort if you need ordering.
sort	field name / list	Sort rows by the given fields.
select	Jinja expression	Keep only rows where the expression is truthy (select: row.is_active).
cut	field name / list	Keep only the listed fields.
cutout	field name / list	Drop the listed fields.
head	integer	Keep only the first N rows.

RowmapConfig:

Field	Type	Required	Description
iterable	RowmapIterable	Yes*	The array (or delimited string) to explode. Required for rowmapmany; forbidden for rowmap.
columns	dict[str, str]	Yes	Output column → Jinja expression. Header order follows the dict insertion order.
skip_if	string	No	Jinja expression; rows (or items) matching it are dropped.

RowmapIterable:

Field	Type	Required	Default	Description
field	string	Yes	—	Jinja expression resolving to the iterable, e.g. row.userGroups.
type	string	No	array	array (list of objects/scalars) or split_string (comma-delimited string).
delimiter	string	No	,	Delimiter used when type: split_string.
strip	bool	No	true	Strip whitespace from items when type: split_string.

Expressions are Jinja expressions (no {{ }} braces) and preserve native types: an integer stays an integer, a list stays a list, and a missing value resolves to None. Rows are passed as PETL Record objects, so row.field works for most field names; field names that collide with a Record method (e.g. count) need subscript access (row['count']).

These transforms are lazy — the stream is only consumed as downstream processing pulls it.

Why Not a Second API Call?

Making a second fetch() call instead of using source_type: source has several downsides:

Problem	Impact
Redundant credentials	Requires separate auth, potentially failing if credentials don’t work twice
Memory overhead	Both sources load independently into memory instead of streaming together
Sync issues	Data inconsistencies between two separate API calls (e.g., if API state changes mid-run)
Performance	Each API call has latency, multiplied by pagination overhead
Complexity	More code to maintain, more error handling needed

Real-World Example: BongaWonga Users → Roles

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: bongawonga_api
    params:
      resource: /api/users
    field_mapping:
      id:
        destination_field: unique_id
      email:
        destination_field: email
      name:
        destination_field: name
      role_name:
        destination_field: assigned_role
  
  roles:
    destination: LocalRole
    source_type: source
    source: users
    field_mapping:
      assigned_role:
        destination_field: name
      assigned_role_id:                 # Note: different key to avoid duplicate
        destination_field: unique_id

With a PRE_TRANSFORM hook to extract unique roles:

@hook(event=OAAHookEvent.PRE_TRANSFORM, target="roles")
def extract_unique_roles(source=None, **kwargs):
    """Extract unique role names from user records."""
    def gen_roles():
        seen = set()
        for record in source.stream.dicts():
            role = record.get("assigned_role")
            if role and role not in seen:
                yield {"assigned_role": role}
                seen.add(role)
    
    # Reshape stream to only contain one record per unique role
    source._stream = source.stream.rowmapmany(
        lambda r: [{"assigned_role": r.get("assigned_role")}] if r.get("assigned_role") and r.get("assigned_role") not in set() else [],
        header=["assigned_role"]
    )

tree

Beta: Tree functionality is still in beta and may change in future releases.

Used when a source represents a hierarchical resource (e.g. org chart).

Field	Type	Required	Description
id_field	string	Yes	Field name containing the node’s own ID.
parent_id_field	string	Yes	Field name containing the parent node’s ID.
sub_fieldname	string	No	Name to use for the child list. Defaults to sub_resources.

sources:
  org_chart:
    destination: Resource
    resource_type: Department
    source_type: custom
    connection: my_api
    tree:
      id_field: id
      parent_id_field: manager_id
      sub_fieldname: direct_reports

generators

Declares one or more inline object generators on a source. Each generator reads a field from every source record and automatically creates LocalGroup or LocalRole objects from the unique values it finds — optionally assigning membership at the same time.

Generators are deduplicated automatically: if 50 user records all share the same field value, exactly one group or role is created.

For a full walk-through see How to Use Generators.

Each key under generators is an arbitrary generator name. The value is:

Field	Type	Required	Default	Description
generator_type	string	Yes	—	Type of object to create: local_group or local_role.
source_field	string	Yes	—	Field name to read values from. Must match the destination field name after field_mapping transforms have run.
name_template	string	No	{{value}}	Jinja2 template for the generated object’s name. Use {{value}} for the field value, {{row}} for the full record.
unique_id_template	string	No	{{value}}	Jinja2 template for the generated object’s unique_id.
add_membership	bool	No	true	When true, links the source record (e.g. LocalUser) to the generated object via add_group() or add_role().
skip_empty	bool	No	true	Skip records where source_field is blank or null.
multi_value	bool	No	false	Treat source_field as a Python list and process each element independently.
create_if_missing	bool	No	true	When false (link-only mode), never create new objects — only assign membership to objects that already exist. Objects not found are skipped with a warning.
field_mapping	dict	No	—	Optional extra property mappings to set on the generated object. Only convert-mode transforms are supported here.

Minimal example — only required fields, all other keys take their defaults:

sources:
  users:
    destination: LocalUser
    source_type: csv
    csv_filepath: data/users.csv
    field_mapping:
      department:
        destination_field: department
    generators:
      dept_groups:
        generator_type: local_group
        source_field: department

Fully-specified example — all keys explicit:

sources:
  users:
    destination: LocalUser
    source_type: csv
    csv_filepath: data/users.csv
    field_mapping:
      department:
        destination_field: department
      job_title:
        destination_field: job_title
    generators:
      dept_groups:
        generator_type: local_group
        source_field: department
        name_template: "{{value}}"
        unique_id_template: "dept_{{value}}"
        add_membership: true
        skip_empty: true
        multi_value: false
        create_if_missing: true
      title_roles:
        generator_type: local_role
        source_field: job_title
        name_template: "{{value}}"
        unique_id_template: "role_{{value}}"
        add_membership: true
        skip_empty: true

Link-only mode with a list field — roles already created by a separate source; generator only wires up per-user membership:

sources:
  roles:
    destination: LocalRole
    source_type: custom
    connection: api
    params:
      resource: /roles
    field_mapping:
      id:
        destination_field: unique_id
      name:
        destination_field: name

  users:
    destination: LocalUser
    source_type: custom
    connection: api
    params:
      resource: /users
    field_mapping:
      id:
        destination_field: unique_id
      role_ids:
        destination_field: role_ids
    generators:
      user_roles:
        generator_type: local_role
        source_field: role_ids
        unique_id_template: "{{value}}"
        multi_value: true          # role_ids is a list
        create_if_missing: false   # roles already exist from the roles source above
        add_membership: true

Source ordering with create_if_missing: false: Sources are processed in the order they are declared. When using link-only mode, the source that creates the target objects must be declared before the source whose generator references them.

---

connections

Each key under connections is a named connection. The value’s shape depends on connection_type.

All connection types support the required_env field:

Field	Type	Required	Description
required_env	list[string]	No	List of environment variable names that this connection requires. oaa-runner validates that each variable is set before running, and auto-loads them from .env and injects them into connection.params. See Required environment variables for details.

CSV connection

connections:
  my_csv:
    connection_type: csv

No additional fields required. The file path is set on the source via csv_filepath.

API connection

Field	Type	Required	Description
connection_type	string	Yes	api
uri	string	Yes	Base URI of the API.
fetch_module	string	Yes	Python module that implements fetch().
method	string	No	HTTP method: GET (default) or POST.
params	dict	No	Default query parameters.
auth	dict	No	Authentication parameters passed to fetch_module.
max_results	int	No	Maximum number of records to fetch.
required_env	list[string]	No	Environment variables required by this connection.

connections:
  my_api:
    connection_type: api
    uri: https://api.example.com
    fetch_module: modules.myapp
    method: GET
    required_env:
      - MY_API_KEY
      - MY_API_SECRET
    auth:
      api_token: "!env"

Veza API connection

Field	Type	Required	Description
connection_type	string	Yes	veza_api
path	string	Yes	Veza API path to query (e.g. /api/v1/users).
response_type	string	No	Expected response shape: list (default).
max_results	int	No	Maximum number of records to fetch.
required_env	list[string]	No	Environment variables required by this connection.

connections:
  veza:
    connection_type: veza_api
    path: /api/v1/resources
    response_type: list
    max_results: 1000

DB connection

Field	Type	Required	Description
connection_type	string	Yes	db
DB_DIALECT	string	Yes	Database type: SQLSERVER, SQLSERVER.ODBC, ORACLE, POSTGRES.
DB_SERVER	string	Yes	Hostname or IP of the database server.
DB_NAME	string	Yes	Database name.
DB_USER	string	Yes	Database username. Use !env to read from an environment variable of the same name.
DB_PASSWORD	string	Yes	Database password. Use !env to read from an environment variable of the same name. Defaults to !env.
DB_PORT	int	No	Database port.
QUERY	string	Yes*	SQL query string. Required if QUERY_PATH and QUERY_STREAMS are not set.
QUERY_PATH	string	Yes*	Path to a .sql file. Required if QUERY and QUERY_STREAMS are not set.
QUERY_STREAMS	list	Yes*	Named query streams. Required if QUERY and QUERY_PATH are not set.
required_env	list[string]	No	Environment variables required by this connection.

connections:
  my_db:
    connection_type: db
    DB_DIALECT: SQLSERVER
    DB_SERVER: db.example.com
    DB_NAME: mydb
    DB_USER: sa
    DB_PASSWORD: "!env"
    QUERY_PATH: ./queries/users.sql

Custom connection

Field	Type	Required	Description
connection_type	string	Yes	custom
custom_module	string	Yes	Python module name that implements fetch(connection, source, **kwargs). Looked up first as-is, then as oaa.modules.<name>.
query	string	No	Optional query string passed to the module.
params	dict	No	Arbitrary parameters passed to fetch().
required_env	list[string]	No	Environment variables required by this connection.

connections:
  my_custom:
    connection_type: custom
    custom_module: modules.myapi
    required_env:
      - MY_API_KEY
      - MY_API_SECRET
    params:
      base_url: https://api.example.com

Common connection

No-code connection type. Delegates all fetching to the concurrent_fetch built-in hook. No fetch_module required.

Field	Type	Required	Description
connection_type	string	Yes	common
common_hooks	list	Yes	List of built-in hook configs. Must include concurrent_fetch.
required_env	list[string]	No	Environment variables required by this connection.

connections:
  api:
    connection_type: common
    required_env: [MY_CLIENT_ID, MY_CLIENT_SECRET]
    common_hooks:
      - name: concurrent_fetch
        options:
          base_url: "https://api.example.com"
          auth:
            type: oauth_client_credentials
            client_id: "${MY_CLIENT_ID}"
            client_secret: "${MY_CLIENT_SECRET}"

See how-to-enable-common-hooks.md for full options.

---

permission_to_veza_map

Maps application permission names to Veza permission types.

permission_to_veza_map:
  read:
    permissions:
      - DataRead
  write:
    name: Write Access
    permissions:
      - DataWrite
      - DataRead

Field	Type	Required	Description
name	string	No	Display name for this permission mapping.
permissions	list[string]	Yes	List of Veza OAAPermission values (e.g. DataRead, DataWrite, NonData).

Migration note: Earlier versions accepted a bare list as the value (e.g. read: [DataRead]). This format is no longer supported. Wrap the list in a permissions key as shown above. oaa-runner will raise a clear error if the old format is detected.

Valid Veza Canonical Permissions

Admin is not a valid permission value. Use the full permission set:

Permission	Purpose
DataRead	Read/view data
DataWrite	Modify data
DataCreate	Create new data
DataDelete	Delete data
MetadataRead	Read metadata/configuration
MetadataWrite	Modify metadata
MetadataCreate	Create new metadata items
MetadataDelete	Delete metadata items

Correct format (with permissions: wrapper):

permission_to_veza_map:
  Admin:
    permissions:
      - DataRead
      - DataWrite
      - DataCreate
      - DataDelete
      - MetadataRead
      - MetadataWrite
      - MetadataCreate
      - MetadataDelete
  Editor:
    permissions:
      - DataRead
      - DataWrite
  Viewer:
    permissions:
      - DataRead

Incorrect (bare list, no permissions: wrapper):

permission_to_veza_map:
  Admin:  # ❌ Missing permissions: wrapper
    - DataRead

Incorrect (Admin value):

permission_to_veza_map:
  Admin:
    permissions:
      - Admin  # ❌ Not a valid permission value

---

run_params

Runtime execution flags that control how a single run behaves. These can also be overridden by CLI arguments (CLI takes precedence).

run_params:
  dry_run: false       # parse data but don't push to Veza
  save_json: false     # save payload JSON to local file
  no_push: false       # skip all Veza API calls entirely
  priority_push: false # push with priority flag

Field	Type	Required	Default	Description
dry_run	bool	No	false	Parse data but don’t push to Veza.
save_json	bool	No	false	Save payload JSON to a local file.
no_push	bool	No	false	Skip all Veza API calls entirely.
priority_push	bool	No	false	Push with priority flag.

---

Common hooks

common_hooks enable framework-owned, reusable behaviors by name — the same idea as transforms, but for whole-stream / lifecycle / fetch behavior. Each entry is a CommonHook: either a bare hook name (string) or a name + options mapping.

common_hooks:
  - filter_stream_by_field                  # shorthand
  - name: filter_stream_by_field            # with options
    options:
      field: Trustee_Type
      operator: eq
      value: "1"
      action: keep

Declaration levels

A common hook can be declared at three levels, which set what it is scoped to:

Level	Where	Scope
Global	top-level common_hooks:	the whole run (every source)
Source	sources.<name>.common_hooks	that one source
Connection	connections.<name>.common_hooks	every source that uses the connection

Lifecycle hooks bind to a lifecycle event (e.g. PRE_TRANSFORM) that the hook itself declares — config only enables the hook and supplies options.

Built-in common hooks

Name	Phase	Event	Description
concurrent_fetch	fetch	—	Concurrent offset-paginated fetch. Declared on a connection (or source). See How to Enable Common Hooks.
credential_store	lifecycle	PRE_RUN	Resolve a secret from an external credential store into an env_var before the run starts. Skips the backend call if the env var is already set. Options: backend, env_var, plus backend-specific keys (all support ${ENV_VAR} interpolation). See How to Enable Common Hooks.
filter_stream_by_field	lifecycle	PRE_TRANSFORM	Keep or drop source rows by a field value. Options: field (raw source field), operator (eq/ne/in/contains/startswith/regex), value, action (keep/drop).
hello_world	lifecycle	PRE_TRANSFORM	Prints a message per scoped source — a trivial hook for smoke-testing common-hook wiring. Option: message (default hello world).
sqlite_streaming_push	push	—	Memory-bounded push for large populations: serializes the provider to SQLite one object at a time, frees the in-memory graph, then streams the payload from disk into a multipart upload — avoiding the full-dict + full-JSON-string peak of the default push. Declared globally. Options: free_memory (default true), part_size (bytes, default 50 MB), db_path (default temp), keep_files (default false). v1 supports CustomApplication providers.

# Source-scoped filter: keep only rows where kind == "user"
sources:
  users:
    source_type: custom
    connection: api
    common_hooks:
      - name: filter_stream_by_field
        options:
          field: kind
          operator: eq
          value: user
          action: keep

---

Required environment variables

The required_env field on any connection explicitly declares which environment variables are required for that connection. It has three behaviors:

1. Validation — Before running, oaa-runner checks that every listed variable is set in the environment (or .env file). If any are missing, the run fails with a clear error naming the variable and the connection that declared it.

2. Auto-loading — .env files are loaded from the same directory as the config.yaml file. When required_env is used, integration modules no longer need to call load_dotenv() themselves.

3. Injection — Values for listed variables are automatically injected into connection.params, so integration code can access them as connection.params['MY_API_KEY'] instead of calling params_or_env().

VEZA_URL and VEZA_API_KEY are always implicitly required regardless of whether required_env is present — they do not need to be listed explicitly.

connections:
  my_api:
    connection_type: custom
    custom_module: modules.myapp
    required_env:
      - MY_API_KEY
      - MY_API_SECRET
    params:
      base_url: https://api.example.com

With the above config and a .env file containing MY_API_KEY=abc123, the integration’s fetch() function receives connection.params as:

{
    "base_url": "https://api.example.com",   # from config
    "MY_API_KEY": "abc123",                   # auto-loaded from .env
    "MY_API_SECRET": "def456",                # auto-loaded from .env
}

---

Environment variable substitution

There are two ways to pull environment values into the config, plus the required_env injection described above.

${VAR} interpolation (config-wide)

Any string value anywhere in config.yaml may reference an environment variable with ${VAR} syntax. References are resolved once, at config-load time, across the entire file — app, sources, connections, hook options, everything:

app:
  name: dcps-${ENVIRONMENT_NAME}
  application_type: dcps-${ENVIRONMENT_NAME}

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}"
          auth:
            type: oauth_client_credentials
            client_id: "${DCPS_CLIENT_ID}"
            client_secret: "${DCPS_CLIENT_SECRET}"

Behavior:

• .env files are loaded automatically (from both the current directory and the config.yaml directory) before interpolation, so variables defined there are available. Real environment variables take precedence over .env.

• If a referenced variable is not set, the ${VAR} text is left unchanged and a warning is logged — a non-env ${...} literal is never silently blanked out.

• A ${VAR} substring is replaced in place, so dcps-${ENVIRONMENT_NAME} becomes dcps-prod.

Use required_env on a connection (below) when you want a run to fail fast if a variable is missing rather than fall back to the literal.

!env substitution (field-name match)

Fields with the exact value !env are replaced at runtime with the value of the environment variable whose name matches the field name:

connections:
  my_db:
    connection_type: db
    DB_DIALECT: SQLSERVER
    DB_USER: sa
    DB_PASSWORD: "!env"   # reads DB_PASSWORD from environment

This differs from ${VAR}: !env matches the field’s own name and replaces the whole value, whereas ${VAR} can name any variable and can be embedded inside a larger string.

For connections with required_env, values are additionally injected into connection.params — see Required environment variables.

---

Validating your config

Before running a sync, validate the config file:

oaa app validate --config=config.yaml

# Also test live connectivity to data sources:
oaa app validate --config=config.yaml --test-connectivity