Ci

CI/CD convenience functions for flow deployment

CI convenience functions for NiFi flow management.

These functions wrap nipyapi operations with: - Environment variable support for CI/CD platforms - Sensible defaults - Simplified interfaces

They return plain dicts and raise exceptions on error. The CLI handles output formatting, logging capture, and exit codes.

Example:

nipyapi ci ensure_registry --repo owner/repo
nipyapi ci deploy_flow --bucket flows --flow my-flow
nipyapi ci start_flow
Profile resolution:

CI functions auto-detect configuration source: 1. If NIFI_API_ENDPOINT env var is set, use environment variables 2. Otherwise, use first profile from ~/.nipyapi/profiles.yml if it exists 3. Fall back to development examples if neither available

nipyapi.ci.change_flow_version(process_group_id: str | None = None, target_version: str | None = None, branch: str | None = None, token: str | None = None, repo: str | None = None, provider: str | None = None) → dict[source]

Change the version of a deployed flow.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • target_version – Version to change to (commit SHA, tag, or branch name). Env: NIFI_TARGET_VERSION. If None, changes to latest.

  • branch – Branch to use. Env: NIFI_FLOW_BRANCH

  • token – Git token for resolving tags. Env: GH_REGISTRY_TOKEN or GL_REGISTRY_TOKEN

  • repo – Repository in owner/repo format. Env: NIFI_REGISTRY_REPO

  • provider – Git provider (github/gitlab). Env: NIFI_REGISTRY_PROVIDER

Returns:

dict with previous_version, new_version, version_state

Raises:

ValueError – Missing required parameters or not under version control

nipyapi.ci.cleanup(process_group_id: str | None = None, stop_only: bool | None = None, force: bool | None = None, delete_parameter_context: bool | None = None, delete_orphaned_contexts: bool | None = None, disable_controllers: bool | None = None) → dict[source]

Stop and optionally delete a process group.

By default, this function stops the process group and disables its controllers, but does NOT delete it or its parameter context. Use explicit flags or environment variables for destructive operations.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • stop_only – Only stop processors, don’t delete anything. Env: NIFI_STOP_ONLY (default: false)

  • force – Force deletion even with queued FlowFiles. Env: NIFI_FORCE_DELETE (default: false)

  • delete_parameter_context – Also delete the parameter context. Env: NIFI_DELETE_PARAMETER_CONTEXT (default: false) WARNING: Only use if you’re certain no other process groups share this context (e.g., Openflow connectors share contexts).

  • delete_orphaned_contexts – After deletion, also delete any parameter contexts that are no longer bound to any process groups. Env: NIFI_DELETE_ORPHANED_CONTEXTS (default: false) This is safer than delete_parameter_context as it only removes contexts that are definitely unused.

  • disable_controllers – Disable controller services after stopping. Env: NIFI_DISABLE_CONTROLLERS (default: true)

Returns:

dict with stopped/deleted status and names

Raises:

ValueError – Missing required parameters

Example:

# Just stop the flow (safest)
nipyapi ci cleanup --process_group_id PG_ID --stop_only

# Stop and delete process group only (safe for shared contexts)
nipyapi ci cleanup --process_group_id PG_ID

# Full cleanup including parameter context (CI/CD pipelines)
nipyapi ci cleanup --process_group_id PG_ID --delete_parameter_context --force

# Clean up orphaned contexts after deletion (safe)
nipyapi ci cleanup --process_group_id PG_ID --delete_orphaned_contexts

# Via environment variables (for CI/CD)
NIFI_DELETE_PARAMETER_CONTEXT=true NIFI_FORCE_DELETE=true nipyapi ci cleanup
nipyapi.ci.commit_flow(process_group_id: str | None = None, comment: str | None = None, registry_client: str | None = None, bucket: str | None = None, flow_name: str | None = None, force: bool = False) → dict[source]

Commit a flow to version control.

For flows already under version control, saves a new version with the provided comment. For new flows, starts version control by saving the initial version to the specified registry client and bucket.

This function uses nipyapi.versioning.save_git_flow_ver for Git-based registries (GitHub, GitLab, etc.).

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • comment – Commit message describing the changes. Env: NIFI_COMMIT_COMMENT

  • registry_client – Registry client name (for initial commit only). Env: NIFI_REGISTRY_CLIENT

  • bucket – Bucket/folder name (for initial commit only). Env: NIFI_BUCKET

  • flow_name – Name for the flow in registry (for initial commit only). Defaults to process group name. Env: NIFI_FLOW_NAME

  • force – If True, use FORCE_COMMIT to ignore merge conflicts.

Returns:

  • flow_id: Flow identifier in registry

  • version: Commit SHA of the new version

  • state: Version control state after commit

  • initial_commit: True if this was the first commit

Return type:

dict with

Raises:

ValueError – Missing required parameters or registry not found

nipyapi.ci.configure_inherited_params(process_group_id: str | None = None, parameters: str | Dict[str, Any] | None = None, parameters_file: str | None = None, dry_run: bool = False, allow_override: bool = False) → dict[source]

Configure parameters in their owning contexts within an inheritance hierarchy.

This function determines which parameter context owns each parameter and updates it in the correct context, preventing accidental shadowing.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • parameters – JSON string or dict of parameter name -> value pairs. Env: NIFI_PARAMETERS

  • parameters_file – Path to JSON or YAML file with parameter name -> value pairs. Env: NIFI_PARAMETERS_FILE

  • dry_run – If True, return the plan without making changes. Env: NIFI_DRY_RUN (default: false)

  • allow_override – If True, create parameters at top level even if they exist in inherited contexts. Env: NIFI_ALLOW_OVERRIDE (default: false)

Returns:

dry_run, plan, parameters_updated, contexts_modified, warnings, and errors.

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or invalid JSON

Example:

# From inline JSON
nipyapi ci configure_inherited_params \
    --process_group_id abc123 \
    --parameters '{"PostgreSQL Username": "myuser"}'

# From file
nipyapi ci configure_inherited_params \
    --process_group_id abc123 \
    --parameters_file params.yaml

# Dry run
nipyapi ci configure_inherited_params \
    --process_group_id abc123 \
    --parameters_file params.json \
    --dry_run
nipyapi.ci.configure_params(process_group_id: str | None = None, parameters: str | Dict[str, Any] | None = None) → dict[source]

Configure parameters on a process group’s parameter context.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • parameters – JSON string or dict of parameter name -> value pairs. Env: NIFI_PARAMETERS

Returns:

dict with parameters_updated, parameters_count, context_name

Raises:

ValueError – Missing required parameters or invalid JSON

nipyapi.ci.delete_nar(identifier: str | None = None, group: str | None = None, artifact: str | None = None, version: str | None = None, force: bool | None = None) → dict[source]

Delete an installed NAR extension from NiFi.

Can specify NAR by identifier or by Maven coordinate (group:artifact:version).

Parameters:

  • identifier – NAR identifier (UUID). Env: NIFI_NAR_ID

  • group – Maven group ID (alternative to identifier). Env: NIFI_NAR_GROUP

  • artifact – Maven artifact ID (required with group). Env: NIFI_NAR_ARTIFACT

  • version – Maven version (required with group). Env: NIFI_NAR_VERSION

  • force – Force deletion even if components are in use. Env: NIFI_NAR_FORCE. Defaults to False.

Returns:

  • deleted: “true” if deletion succeeded

  • identifier: The deleted NAR’s identifier

  • group: Maven group ID

  • artifact: Maven artifact ID

  • version: Maven version

Return type:

dict with

Raises:

ValueError – Missing required parameters or NAR not found

Example:

# Delete by identifier
nipyapi ci delete_nar --identifier abc-123-def

# Delete by coordinate
nipyapi ci delete_nar --group com.example --artifact my-nar --version 1.0.0

# Force delete
nipyapi ci delete_nar --identifier abc-123-def --force
nipyapi.ci.deploy_flow(registry_client: str | None = None, bucket: str | None = None, flow: str | None = None, parent_id: str | None = None, branch: str | None = None, version: str | None = None, location: Tuple[int, int] | None = None, greedy: bool = False, parameter_context_handling: str | None = None) → dict[source]

Deploy a flow from a Git-based registry to NiFi.

Parameters:

  • registry_client – Registry client ID (UUID) or name. Auto-detects which based on format. Env: NIFI_REGISTRY_CLIENT_ID

  • bucket – Bucket (folder) containing the flow. Env: NIFI_BUCKET

  • flow – Flow name (filename without .json). Env: NIFI_FLOW

  • parent_id – Parent Process Group ID. Env: NIFI_PARENT_ID (default: root)

  • branch – Branch to deploy from. Env: NIFI_FLOW_BRANCH

  • version – Version (commit SHA, tag, branch). Env: NIFI_TARGET_VERSION

  • location – (x, y) tuple for placement on canvas

  • greedy – If True, allow partial name matching for registry_client. Default False (exact match for safety in CI/automation).

  • parameter_context_handling – How to handle existing parameter contexts with the same name. Valid values: - ‘KEEP_EXISTING’: Reuse existing context by name (default) - ‘REPLACE’: Create new context with numbered suffix Env: NIFI_PARAMETER_CONTEXT_HANDLING

Returns:

dict with process_group_id, process_group_name, deployed_version

Raises:

ValueError – Missing required parameters or registry client not found

nipyapi.ci.detach_flow(process_group_id: str | None = None) → dict[source]

Detach a flow from version control.

Removes version control from a process group, allowing it to be re-versioned to a different registry. This is the first step in forking a read-only flow to your own repository.

After detaching, use commit_flow to save the flow to your own registry client and bucket.

Parameters:

process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

Returns:

  • detached: True if successfully detached

  • process_group_name: Name of the process group

  • previous_registry: Name of the previous registry (if available)

  • previous_flow_id: Previous flow identifier

Return type:

dict with

Raises:

ValueError – Missing required parameters or not under version control

Example:

# Fork workflow: detach from read-only registry, save to your own
nipyapi ci detach_flow --process_group_id PG_ID
nipyapi ci commit_flow --process_group_id PG_ID --registry_client MyRepo --bucket flows
nipyapi.ci.ensure_registry(token: str | None = None, repo: str | None = None, client_name: str | None = None, provider: str | None = None, api_url: str | None = None, default_branch: str | None = None, repository_path: str | None = None) → dict[source]

Ensure a Git Flow Registry Client exists with the desired configuration.

Parameters:

  • token – Personal Access Token. Env: GH_REGISTRY_TOKEN or GL_REGISTRY_TOKEN

  • repo – Repository in owner/repo format. Env: NIFI_REGISTRY_REPO

  • client_name – Registry client name. Env: NIFI_REGISTRY_CLIENT_NAME

  • provider – “github” or “gitlab”. Env: NIFI_REGISTRY_PROVIDER

  • api_url – API URL override. Env: NIFI_REGISTRY_API_URL

  • default_branch – Default branch. Env: NIFI_REGISTRY_BRANCH

  • repository_path – Path in repo. Env: NIFI_REPOSITORY_PATH

Returns:

dict with registry_client_id and registry_client_name

Raises:
nipyapi.ci.export_flow_definition(process_group_id: str | None = None, file_path: str | None = None, mode: str | None = None, include_referenced_services: bool | None = None) → dict[source]

Export a process group as a flow definition file.

Exports the current state of a process group to a JSON or YAML file. This does NOT require NiFi Registry - it exports the live canvas state.

Parameters:

  • process_group_id – ID of the process group to export. Env: NIFI_PROCESS_GROUP_ID

  • file_path – Path to write the flow definition to. Env: NIFI_FLOW_FILE_PATH If not provided, returns the flow definition as a string in the result.

  • mode – Export format - ‘json’ or ‘yaml’. Defaults to ‘json’. Env: NIFI_FLOW_FORMAT

  • include_referenced_services – If True, include controller services from outside the target group that are referenced by components within. Env: NIFI_INCLUDE_REFERENCED_SERVICES. Defaults to False.

Returns:

process_group_id, process_group_name, file_path, format, and flow_definition (if file_path not provided).

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or process group not found

Example:

# Export to file
nipyapi ci export_flow_definition \
    --process_group_id abc123 \
    --file_path my-flow.json

# Export to stdout (for piping)
nipyapi ci export_flow_definition --process_group_id abc123

# Export with external controller services included
nipyapi ci export_flow_definition \
    --process_group_id abc123 \
    --include_referenced_services
nipyapi.ci.export_parameters(context_id: str | None = None, process_group_id: str | None = None, file_path: str | None = None, mode: str = 'json', include_hierarchy: bool | None = None) → dict[source]

Export parameters from a parameter context hierarchy.

By default, exports a flat {name: value} dict suitable for import via configure_inherited_params. Optionally includes the full hierarchy structure.

Note: Sensitive parameter values cannot be exported (NiFi returns null), but their keys are included so users know what needs to be recreated.

Parameters:

  • context_id – ID of the parameter context. Env: NIFI_PARAMETER_CONTEXT_ID

  • process_group_id – Optional. Resolve context from this PG instead. Env: NIFI_PROCESS_GROUP_ID

  • file_path – Optional path to write the export. Env: NIFI_PARAMETERS_FILE If not provided, returns parameters in the result dict.

  • mode – Output format: ‘json’ or ‘yaml’. Env: NIFI_EXPORT_MODE (default: json)

  • include_hierarchy – If True, include full hierarchy structure showing which context owns each parameter. Env: NIFI_INCLUDE_HIERARCHY

Returns:

context_id, context_name, file_path, format, parameter_count, and parameters (if file_path not provided).

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or context not found

Example:

# Export by context ID
nipyapi ci export_parameters --context_id abc123

# Export from process group's context
nipyapi ci export_parameters --process_group_id xyz789

# Export to file
nipyapi ci export_parameters --context_id abc123 \
    --file_path params.yaml --mode yaml
nipyapi.ci.get_flow_diff(process_group_id: str | None = None) → dict[source]

Get local modifications to a versioned flow.

Returns structured information about all local changes made to a version-controlled process group since the last commit/sync.

Use this before upgrading a flow to capture modifications that need to be re-applied after the upgrade.

This function uses nipyapi.versioning.get_local_modifications for the underlying API call.

Parameters:

process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

Returns:

  • process_group_id: ID of the process group

  • process_group_name: Name of the process group

  • flow_id: Flow ID in registry

  • current_version: Current committed version

  • state: Version control state

  • modification_count: Number of modified components

  • modifications: List of modification details

Return type:

dict with

Raises:

ValueError – Missing required parameters or not under version control

nipyapi.ci.get_flow_versions(process_group_id: str | None = None) → dict[source]

Get version history for a versioned flow.

Lists all versions (commits) of a flow with their metadata including commit SHA, author, timestamp, and comments.

Parameters:

process_group_id – ID of the versioned process group. Env: NIFI_PROCESS_GROUP_ID

Returns:

  • flow_id: Flow identifier in registry

  • bucket_id: Bucket containing the flow

  • registry_id: Registry client ID

  • current_version: Currently deployed version

  • versions: List of version metadata

Return type:

dict with

Raises:

ValueError – Missing required parameters or not under version control

nipyapi.ci.get_status(process_group_id: str | None = None) → dict[source]

Get comprehensive status information for a process group.

Parameters:

process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID. If not provided, defaults to root process group.

Returns:

  • process_group_id, process_group_name, state, is_root

  • Processor counts (total, running, stopped, invalid, disabled)

  • Controller counts (total, enabled, disabled)

  • Version control info (versioned, version_id, version_state, etc.)

  • Parameter context info

  • Bulletin counts (warnings/errors)

Return type:

dict with status information including

Example:

# Get status of specific process group
nipyapi ci get_status --process_group_id PG_ID

# Get status of entire canvas (root process group)
nipyapi ci get_status
nipyapi.ci.import_flow_definition(file_path: str | None = None, flow_definition: str | None = None, parent_id: str | None = None, location: Tuple[int, int] | None = None) → dict[source]

Import a flow definition file as a new process group.

Creates a new process group from a JSON or YAML flow definition file. This does NOT require NiFi Registry - it imports directly to the canvas.

Parameters:

  • file_path – Path to the flow definition file (JSON or YAML). Env: NIFI_FLOW_FILE_PATH

  • flow_definition – Flow definition as a string (alternative to file_path). Env: NIFI_FLOW_DEFINITION

  • parent_id – Parent process group ID to import into. Env: NIFI_PARENT_ID Defaults to root process group.

  • location – (x, y) tuple for placement on canvas. Env: NIFI_LOCATION_X, NIFI_LOCATION_Y If not provided, uses smart positioning to find a non-overlapping location in a grid layout.

Returns:

process_group_id, process_group_name, parent_id, and source (“file” or “string”).

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or invalid flow definition

Example:

# Import from file
nipyapi ci import_flow_definition \
    --file_path my-flow.json \
    --parent_id abc123

# Import from stdin (for piping)
cat my-flow.json | nipyapi ci import_flow_definition --flow_definition "$(cat)"
nipyapi.ci.list_flows(process_group_id: str | None = None, descendants: bool | None = None) → dict[source]

List flows (process groups) with their version control state.

Lists flows under the specified parent (or root) with their version control state, making it easy to see which flows need updates.

Parameters:

  • process_group_id – Parent process group ID to start from. Env: NIFI_PROCESS_GROUP_ID. If not provided, defaults to root process group.

  • descendants – If True, include all nested process groups recursively. If False (default), only show immediate children. Env: NIFI_INCLUDE_DESCENDANTS.

Returns:

  • parent_id: The parent process group ID searched from

  • parent_name: The parent process group name

  • total_count: Total number of process groups found

  • versioned_count: Number of version-controlled process groups

  • stale_count: Number of process groups with available updates

  • modified_count: Number of locally modified process groups

  • process_groups: List of dicts with version info for each PG

Return type:

dict with

Each process_group entry contains:

  • id, name: Process group identification

  • versioned: Whether under version control

  • current_version: Current version (or empty)

  • state: Version state (UP_TO_DATE, STALE, LOCALLY_MODIFIED, etc.)

  • modified: Whether local changes exist

  • flow_name, bucket_name: Registry flow info

Example:

# List immediate child process groups from root
nipyapi ci list_flows

# List all process groups recursively
nipyapi ci list_flows --descendants

# List process groups under a specific parent
nipyapi ci list_flows --process_group_id PG_ID
nipyapi.ci.list_nars() → dict[source]

List all installed NAR extensions in NiFi.

Returns:

  • count: Number of installed NARs

  • nars: List of NAR summaries with coordinate, state, extensions

Return type:

dict with

Example:

nipyapi ci list_nars
nipyapi.ci.list_registry_flows(registry_client: str | None = None, bucket: str | None = None, branch: str | None = None, detailed: bool = False, greedy: bool = False) → dict[source]

List flows available in a Git registry bucket.

This function queries a Git-based Flow Registry to list available flows within a bucket. Useful for discovering what flows can be deployed.

Parameters:

  • registry_client – Registry client ID (UUID) or name. Auto-detects which based on format. Env: NIFI_REGISTRY_CLIENT_ID

  • bucket – Bucket (folder) containing flows. Env: NIFI_BUCKET

  • branch – Branch to query. Env: NIFI_FLOW_BRANCH. If not specified, uses the registry client’s default branch.

  • detailed – If True, include description and comments for each flow. Default False returns minimal info (name, flow_id).

  • greedy – If True, allow partial name matching for registry_client. Default False (exact match for safety in CI/automation).

Returns:

  • registry_client_id: The resolved registry client ID

  • registry_client_name: The registry client name

  • bucket: The bucket queried

  • branch: The branch queried (if specified)

  • flow_count: Number of flows found

  • flows: List of flow info dicts

Return type:

dict with

Raises:

ValueError – If registry client or bucket not found, or multiple registry clients match when greedy=False.

Example:

# List all flows in bucket
nipyapi ci list_registry_flows --registry_client my-github --bucket flows

# With full details
nipyapi ci list_registry_flows --registry_client my-github --bucket flows --detailed
nipyapi.ci.purge_flowfiles(process_group_id: str | None = None, stop: bool = True) → dict[source]

Purge all queued flow files from a process group.

This drops all FlowFiles from all connections within the process group. Useful for clearing stuck data or recovering from failed states.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • stop – Stop the process group before purging (recommended). Default: True

Returns:

dict with purge results

Raises:

ValueError – Missing required parameters

Example:

# Purge with stop (safest)
nipyapi ci purge_flowfiles --process_group_id PG_ID

# Purge without stopping (use with caution)
nipyapi ci purge_flowfiles --process_group_id PG_ID --stop false
nipyapi.ci.resolve_git_ref(ref: str | None, repo: str | None = None, token: str | None = None, provider: str = 'github') → str | None[source]

Resolve a git ref (tag/branch/SHA) to a commit SHA.

If the ref already looks like a SHA (7-40 hex characters), returns it as-is. Otherwise, calls the GitHub or GitLab API to resolve the ref to a SHA.

Parameters:

  • ref – Tag name, branch name, or commit SHA. If None or empty, returns None.

  • repo – Repository in owner/repo format (GitHub) or namespace/repo (GitLab).

  • token – Personal access token for API access.

  • provider – Git provider - “github” or “gitlab” (default: github).

Returns:

Resolved commit SHA, or None if ref was empty.

Raises:

ValueError – If the ref cannot be resolved.

Example:

>>> resolve_git_ref("v1.0.0", "owner/repo", "ghp_xxx", "github")
"abc123def456..."
>>> resolve_git_ref("abc123def456", None, None)  # Already a SHA
"abc123def456"
nipyapi.ci.revert_flow(process_group_id: str | None = None) → dict[source]

Revert uncommitted changes to a deployed flow.

Restores the process group to match the version tracked in the registry.

Parameters:

process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

Returns:

dict with reverted, version, state

Raises:

ValueError – Missing required parameters or not under version control

nipyapi.ci.start_flow(process_group_id: str | None = None, enable_controllers: bool = True) → dict[source]

Start a process group (enable controllers, start processors).

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • enable_controllers – Enable controller services first (default: True)

Returns:

dict with started, process_group_name

Raises:
nipyapi.ci.stop_flow(process_group_id: str | None = None, disable_controllers: bool | None = None) → dict[source]

Stop a process group (stop processors).

By default, this function only stops processors. Controller services remain enabled so the flow can be quickly restarted.

Use –disable_controllers if you need to delete the process group afterward (deletion requires disabled controllers and purged queues).

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • disable_controllers – Also disable controller services. Env: NIFI_DISABLE_CONTROLLERS (default: false) Only needed before deletion.

Returns:

dict with stopped, process_group_name, controllers_disabled

Raises:

Example:

# Simple stop (processors only, controllers stay enabled)
nipyapi ci stop_flow --process_group_id PG_ID

# Stop for deletion (also disable controllers)
nipyapi ci stop_flow --process_group_id PG_ID --disable_controllers
nipyapi.ci.upload_asset(process_group_id: str | None = None, context_id: str | None = None, file_path: str | None = None, url: str | None = None, filename: str | None = None, param_name: str | None = None) → dict[source]

Upload an asset and optionally link it to a parameter.

This function uploads a file (from local path or URL) as an asset to a parameter context, and can optionally update a parameter to reference it.

Parameters:

  • process_group_id – Process group to find parameter context from. Env: NIFI_PROCESS_GROUP_ID

  • context_id – Direct parameter context ID (alternative to process_group_id). Env: NIFI_PARAMETER_CONTEXT_ID

  • file_path – Path to local file to upload. Env: NIFI_ASSET_FILE_PATH

  • url – URL to download file from (alternative to file_path). Env: NIFI_ASSET_URL

  • filename – Override the filename (defaults to basename from path/URL). Env: NIFI_ASSET_FILENAME

  • param_name – If provided, update this parameter to reference the asset. Env: NIFI_ASSET_PARAM_NAME

Returns:

asset_id, asset_name, asset_digest, context_id, context_name, and parameter_updated.

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or file not found

Example:

# Upload from URL and link to parameter
nipyapi ci upload_asset \
    --url https://jdbc.postgresql.org/download/postgresql-42.7.6.jar \
    --context_id abc123 \
    --param_name "PostgreSQL JDBC Driver"

# Upload from file path
nipyapi ci upload_asset \
    --file_path /path/to/driver.jar \
    --process_group_id def456
nipyapi.ci.upload_nar(file_path: str | None = None, timeout: int | None = None) → dict[source]

Upload a NAR extension file to NiFi and wait for installation.

Parameters:

  • file_path – Path to the NAR file. Env: NIFI_NAR_FILE_PATH

  • timeout – Seconds to wait for installation. Defaults to 120. Env: NIFI_NAR_TIMEOUT

Returns:

  • identifier: NAR identifier

  • group: Maven group ID

  • artifact: Maven artifact ID

  • version: Maven version

  • state: Installation state

  • extension_count: Number of extensions in the NAR

  • processors: List of processor types

  • controller_services: List of controller service types

Return type:

dict with

Raises:

ValueError – Missing required parameters, file not found, or installation fails

Example:

nipyapi ci upload_nar --file_path /path/to/my-nar-1.0.0.nar
nipyapi.ci.verify_config(process_group_id: str | None = None, verify_controllers: bool = True, verify_processors: bool = True) → dict[source]

Verify configuration of all components in a process group.

Validates that all required properties are set and property values meet their defined constraints. Does NOT test actual connectivity or credentials. Designed for CI/CD pipelines to catch configuration errors before starting a flow. Verifies controller services and processors that are in a stopped/disabled state.

Parameters:

  • process_group_id – ID of the process group. Env: NIFI_PROCESS_GROUP_ID

  • verify_controllers – Verify controller services (default: True)

  • verify_processors – Verify processors (default: True)

Returns:

verified (“true”/”false”), failed_count, controller_results, processor_results, summary, and process_group_name. Caller should check verified or failed_count to determine next steps.

Return type:

dict with keys

Raises:

ValueError – Missing required parameters or process group not found

Example:

# CLI usage
nipyapi ci verify_config --process-group-id <pg-id>

# Programmatic usage
result = nipyapi.ci.verify_config(process_group_id)
if result["verified"] == "true":
    nipyapi.ci.start_flow(process_group_id)
else:
    print(f"Verification failed: {result['summary']}")