NiPyAPI Python Client SDK

NiFi prevents the deletion of NARs that are currently being used by active components (processors, controller services, etc.).

Deletion Protection

If you attempt to delete a NAR in use without forcing it, nipyapi.extensions.delete_nar will raise a ValueError with a message indicating components are instantiated from this NAR.

Force Deletion and Orphaned Components

If you use force=True to delete a NAR while components are using it, those components become orphaned. Their state will change to STOPPED or INVALID with a "Missing Processor" error.

Recovering Orphaned Components

To recover components orphaned by a force-deletion, re-upload the exact same NAR version. The components will automatically recover, and their properties and configurations will be preserved.

import nipyapi

# Attempting to delete a NAR in use
try:
    nipyapi.extensions.delete_nar(nar_id)
except ValueError as e:
    # "Unable to delete NAR [...] because components are instantiated from this NAR"
    print(f"NAR protected: {e}")

# Recovering orphaned components by re-uploading
nar = nipyapi.extensions.upload_nar("/path/to/my-processor.nar")

# Wait for processor to recover
proc = nipyapi.extensions.wait_for_processor_init(processor_id)
print(f"Recovered: {proc.component.validation_status}")

When NiFi and the Registry use different certificate authorities, or when you have specific PKI requirements (like Enterprise PKI or component-specific certificates), you must use Explicit SSL Context Services.

To do this:

1. Create an SSL Context Service using nipyapi.security.ensure_ssl_context() providing the keystore and truststore files and passwords.
2. Pass the resulting SSL context object to the ssl_context_service parameter when calling nipyapi.versioning.ensure_registry_client().

# Create SSL Context Service for specific PKI requirements
ssl_context = nipyapi.security.ensure_ssl_context(
    name='registry_ssl_context',
    parent_pg=parent_pg,
    keystore_file='/path/to/specific/keystore.p12',
    keystore_password='password',
    truststore_file='/path/to/specific/truststore.p12',
    truststore_password='password'
)

# Registry Client with explicit SSL Context
registry_client = nipyapi.versioning.ensure_registry_client(
    name='secure_registry_client',
    uri='https://secure-registry.example.com/nifi-registry-api',
    description='Registry Client with explicit SSL Context',
    ssl_context_service=ssl_context
)

NiPyAPI uses Google-style docstrings for both Sphinx documentation and CLI help generation. To ensure compatibility with both:

• Summary: Use triple double-quotes. The first line must be a concise imperative summary.
• Sections: Include Args:, Returns:, and Raises: sections where applicable.
• Types: Do not duplicate type hints in docstrings; focus on semantics and constraints. For union types, use the (type1 or type2) format.
• CLI Compatibility: Do NOT use nested bullet lists under Args parameters, as python-fire will truncate the description in the CLI help text. Use inline formatting instead.
• Examples: Use the Example:: (singular) syntax. You must include a blank line before the code block for Sphinx to render it correctly.
• Cross-references: Use Sphinx notation like :class:~nipyapi.module.ClassName`` to create clickable links. The ~ prefix shows only the class name.

def get_process_group(pg_id, identifier_type='id'):
    """Return a specific process group by identifier.

    Args:
        pg_id (str): The identifier of the process group
        identifier_type (str): 'id' or 'name'

    Returns:
         :class:`~nipyapi.nifi.models.ProcessGroupEntity`: The matching
         process group, or None if not found

    Raises:
        ValueError: If identifier_type is not 'id' or 'name'
    """
    pass

def my_function():
    """Do something useful.

    Example::

        result = my_function()
        print(result)
    """
    pass

Python processors require a unique initialization lifecycle because they must set up a virtual environment and download dependencies before they are ready for configuration.

Initialization States

• initializing: Setting up the Python virtual environment.
• downloading_dependencies: Downloading packages from PyPI.
• dependency_failed: Failed to download dependencies (e.g., network issues).
• ready: Environment is set up, properties are loaded, and the processor is ready for configuration.
• missing_nar: The NAR was removed, leaving the processor orphaned.
• error: An unknown error occurred.

Best Practices

• Air-gapped environments: If NiFi cannot reach PyPI, bundle all dependencies inside the NAR file itself.
• Caching: NiFi caches virtual environments based on the NAR coordinate (group:artifact:version). A version bump (e.g., 1.0.0 to 1.0.1) always triggers a fresh environment creation and full dependency download.

import nipyapi

# Monitor initialization status
status = nipyapi.extensions.get_processor_init_status(processor_id)

print(f"Status: {status['status']}")
print(f"Ready: {status['is_ready']}")
print(f"Has Properties: {status['has_properties']}")
print(f"Message: {status['init_message']}")

if status['validation_errors']:
    for error in status['validation_errors']:
        print(f"  - {error}")