Extensions

NiFi extensions (NAR) management

NiFi Extensions (NAR) Management Module.

Provides functions for managing custom NAR files (extensions) in NiFi. NAR files contain custom processors, controller services, and other components.

NAR Functions:

  • upload_nar() - Upload a NAR file and wait for installation

  • list_nars() - List all installed NARs

  • get_nar() - Get summary of a specific NAR

  • get_nar_details() - Get component types in a NAR

  • get_nar_by_coordinate() - Find NAR by Maven coordinate

  • download_nar() - Download a NAR file

  • delete_nar() - Remove an installed NAR

Processor Initialization Functions:

  • get_processor_init_status() - Get initialization status of extension processor

  • wait_for_processor_init() - Wait for extension processor to initialize

Processor Bundle Version Functions:

  • get_processor_bundle_versions() - List available versions for a processor type

  • get_processor_type_version() - Get processor type for a specific bundle version

  • change_processor_bundle_version() - Change processor to a different bundle version

Python processors and other extension processors may require initialization time after being created (e.g., setting up a virtual environment). Use the processor initialization functions to detect and wait for this.

Example:

>>> import nipyapi
>>> # Upload a NAR and create a processor
>>> nar = nipyapi.extensions.upload_nar('/path/to/custom.nar')
>>> details = nipyapi.extensions.get_nar_details(nar.identifier)
>>> proc_type = nipyapi.canvas.get_processor_type(details.processor_types[0].type)
>>> proc = nipyapi.canvas.create_processor(pg, proc_type, (0,0), 'MyProc')
>>> # Wait for Python processor to initialize
>>> proc = nipyapi.extensions.wait_for_processor_init(proc)
nipyapi.extensions.change_processor_bundle_version(processor, target_version)[source]

Change a processor to use a different bundle version.

When multiple versions of the same processor type are installed (via multiple NARs), this function allows changing which version an existing processor uses. This is different from versioned flow control - it changes the underlying processor implementation.

Note: The processor must be stopped before changing versions.

Parameters:

  • processor – ProcessorEntity or processor ID string

  • target_version (str) – The target bundle version (e.g., ‘0.0.2-SNAPSHOT’) or a BundleDTO object

Returns:

The updated processor

Return type:

ProcessorEntity

Raises:

ValueError – If processor not found, or target version not available

Example:

>>> proc = nipyapi.canvas.get_processor('MyProcessor', 'name')
>>> updated = nipyapi.extensions.change_processor_bundle_version(
...     proc, '0.0.2-SNAPSHOT'
... )
>>> print(f"Now using: {updated.component.bundle.version}")
nipyapi.extensions.delete_nar(identifier, force=False, timeout=30)[source]

Delete an installed NAR from NiFi and wait for cleanup to complete.

When force=True, waits for the NAR to be removed and any processors using it to reach a halting state (deleted, orphaned/missing_nar, or error).

Warning: If force=True and processors from this NAR are still initializing (e.g., Python venv creation in progress), the NiFi API will block. This function checks for initializing processors and raises an error to prevent this. See resources/scripts/NIFI_PYTHON_BRIDGE_BUGS.md for details.

Parameters:

  • identifier (str) – The NAR identifier

  • force (bool) – Force deletion even if components are in use

  • timeout (int) – Maximum seconds to wait for cleanup (default: 30)

Returns:

nar_summary, cleanup_complete, and affected_processors.

Return type:

dict with keys

Raises:

ValueError – If force=True and processors from this NAR are still initializing. Delete the processors first or wait for them to complete initialization.

Example:

>>> result = nipyapi.extensions.delete_nar('abc-123-def', force=True)
>>> if result['cleanup_complete']:
...     print("NAR and all processors cleaned up")
nipyapi.extensions.download_nar(identifier, file_path=None)[source]

Download a NAR file from NiFi.

Parameters:

  • identifier (str) – The NAR identifier

  • file_path (str, optional) – Path to write the NAR file. If None, returns bytes.

Returns:

NAR file contents if file_path is None, otherwise the file path

Return type:

bytes or str

Example:

>>> nipyapi.extensions.download_nar('abc-123', '/tmp/backup.nar')
>>> # Or get bytes directly
>>> nar_bytes = nipyapi.extensions.download_nar('abc-123')
nipyapi.extensions.get_nar(identifier)[source]

Get summary of a specific NAR by its identifier.

Parameters:

identifier (str) – The NAR identifier (UUID)

Returns:

NAR summary object,

or None if not found

Return type:

NarSummaryDTO

Example:

>>> nar = nipyapi.extensions.get_nar('abc-123-def')
>>> print(nar.state)
nipyapi.extensions.get_nar_by_coordinate(group, artifact, version=None)[source]

Find a NAR by its Maven coordinate.

Parameters:

  • group (str) – Maven group ID (e.g., ‘com.example’)

  • artifact (str) – Maven artifact ID (e.g., ‘my-processors-nar’)

  • version (str, optional) – Specific version. If None, returns all versions.

Returns:

Single NAR if version

specified, list of matching NARs otherwise. Returns None if not found.

Return type:

NarSummaryDTO or list

Example:

>>> nar = nipyapi.extensions.get_nar_by_coordinate(
...     'com.example', 'my-processors-nar', '1.0.0'
... )
nipyapi.extensions.get_nar_details(identifier)[source]

Get component types contained in a NAR.

Parameters:

identifier (str) – The NAR identifier (UUID)

Returns:

NAR details with

processor_types, controller_service_types, reporting_task_types. Returns None if not found.

Return type:

NarDetailsEntity

Example:

>>> details = nipyapi.extensions.get_nar_details('abc-123-def')
>>> for proc in details.processor_types:
...     print(proc.type)
nipyapi.extensions.get_processor_bundle_versions(processor_type)[source]

List available bundle versions for a processor type.

When multiple versions of the same processor type are installed (via multiple NARs), this function returns all available bundle versions.

Parameters:

processor_type (str) – The processor type name (e.g., ‘PrepareRegulatoryFile’) or the full qualified type name

Returns:

List of available versions, each with:

  • bundle: BundleDTO with group, artifact, version

  • type: Full processor type name

  • description: Processor description

Return type:

list[dict]

Example:

>>> versions = nipyapi.extensions.get_processor_bundle_versions(
...     'PrepareRegulatoryFile'
... )
>>> for v in versions:
...     print(f"{v['bundle'].version}")
nipyapi.extensions.get_processor_init_status(processor)[source]

Get the initialization status of a processor from a custom NAR.

Python processors and other extension processors may require initialization time after being created (e.g., setting up a virtual environment).

Parameters:

processor – ProcessorEntity or processor ID string

Returns:

  • status: One of ‘initializing’, ‘downloading_dependencies’, ‘dependency_failed’, ‘ready’, ‘missing_nar’, or ‘error’

  • is_ready: True if processor is fully initialized

  • has_properties: True if properties are loaded

  • validation_status: The processor’s validation_status field

  • validation_errors: List of validation error strings

  • init_message: Human-readable description of current state

Return type:

dict with

Example:

>>> status = nipyapi.extensions.get_processor_init_status(proc)
>>> if status['is_ready']:
...     print("Processor ready for configuration")
nipyapi.extensions.get_processor_type_version(processor_type, version)[source]

Get a processor type with a specific bundle version for creation.

When multiple versions of the same processor type are installed (via multiple NARs), use this function to get the correct type for creating a processor with a specific bundle version.

Parameters:

  • processor_type (str) – The processor type name (e.g., ‘PrepareRegulatoryFile’)

  • version (str) – The bundle version (e.g., ‘0.0.2-SNAPSHOT’)

Returns:

The processor type with

the specified bundle version, suitable for passing to nipyapi.canvas.create_processor()

Return type:

DocumentedTypeDTO

Raises:

ValueError – If no matching processor type/version found

Example:

>>> # Get the v2 processor type
>>> proc_type = nipyapi.extensions.get_processor_type_version(
...     'PrepareRegulatoryFile', '0.0.2-SNAPSHOT'
... )
>>> # Create processor with that specific version
>>> proc = nipyapi.canvas.create_processor(pg, proc_type, (0,0), 'MyProc')
nipyapi.extensions.list_nars()[source]

List all installed NARs in NiFi.

Returns:

List of NAR summary objects

Return type:

list[NarSummaryDTO]

Example:

>>> nars = nipyapi.extensions.list_nars()
>>> for nar in nars:
...     print(f"{nar.coordinate.group}:{nar.coordinate.artifact}")
nipyapi.extensions.upload_nar(file_path=None, file_bytes=None, filename=None, timeout=120)[source]

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

This function performs a two-phase wait. First, it waits for the NAR to reach INSTALLED state. Then, for NARs with extensions (especially Python processors), it waits for processor types to be discovered in the API. The discovery phase is important because processor type discovery happens asynchronously after the NAR is marked as installed.

Parameters:

  • file_path (str, optional) – Path to NAR file on disk

  • file_bytes (bytes, optional) – NAR file contents as bytes

  • filename (str, optional) – Filename for the NAR. Required if using file_bytes, defaults to basename of file_path otherwise.

  • timeout (int) – Maximum seconds to wait for both phases (default: 120)

Returns:

Installed NAR summary

Return type:

NarSummaryDTO

Raises:

  • ValueError – If neither file_path nor file_bytes provided, or installation fails

  • FileNotFoundError – If file_path doesn’t exist

Example:

>>> nar = nipyapi.extensions.upload_nar('/path/to/my-nar-1.0.0.nar')
>>> print(f"Installed: {nar.identifier}")
>>> # Processor types are now available
>>> details = nipyapi.extensions.get_nar_details(nar.identifier)
>>> print(f"Processors: {len(details.processor_types)}")
nipyapi.extensions.wait_for_processor_init(processor, timeout=60)[source]

Wait for a processor from a custom NAR to complete initialization.

Python processors need time to set up their virtual environment after being created. This function waits until the processor is ready.

Parameters:

  • processor – ProcessorEntity or processor ID string

  • timeout (int) – Maximum seconds to wait (default: 60)

Returns:

The initialized processor

Return type:

ProcessorEntity

Raises:

ValueError – If processor not found, NAR missing, dependency download failed, or timeout reached

Example:

>>> proc = nipyapi.canvas.create_processor(...)
>>> proc = nipyapi.extensions.wait_for_processor_init(proc)
>>> # Processor is now ready for configuration