Utils

Utility functions and helpers

Convenience utility functions for NiPyApi, not really intended for external use

nipyapi.utils.bypass_slash_encoding(service, bypass)[source]

Instructs the API Client to bypass encoding the ‘/’ character

Parameters:

  • service (str) – ‘nifi’ or ‘registry’

  • bypass (bool) – True will not encode ‘/’ in fields via API calls

Returns:

None

nipyapi.utils.check_version(base, comparator=None, service='nifi', default_version='2.0.0')[source]

Compares version base against either version comparator, or the version of the currently connected service instance.

Since NiFi is java, it may return a version with -SNAPSHOT as part of it. As such, that will be stripped from either the comparator version or the version returned from NiFi

Parameters:

  • base (str) – The base version for the comparison test

  • comparator (optional[str]) – The version to compare against

  • default_version (optional[str]) – The version to assume the service is if the check cannot be completed

  • service (str) – The service to test the version against, currently only supports NiFi

Returns (int): -1/0/1 if base is lower/equal/newer than comparator

Raises:

VersionError – When a feature is not supported in the current version

nipyapi.utils.dump(obj, mode='json')[source]
Dumps a native datatype object or swagger entity to json or yaml

defaults to json

Parameters:

  • obj (varies) – The native datatype object or swagger type to serialise

  • mode (str) – ‘json’ or ‘yaml’, the supported export modes

Returns (str): The serialised object

nipyapi.utils.enforce_min_ver(min_version, bool_response=False, service='nifi')[source]

Raises an error if target NiFi environment is not minimum version.

Parameters:

  • min_version (str) – Version to check against

  • bool_response (bool) – If True, will return True instead of raising error

  • service (str) – nifi or registry

Returns:

(bool) or (NotImplementedError)

nipyapi.utils.exception_handler(status_code=None, response=None)[source]

Simple Function wrapper to handle HTTP Status Exceptions

nipyapi.utils.extract_oidc_user_identity(token_data)[source]

Extract user identity (UUID) from OIDC token response.

This function decodes the JWT access token to extract the ‘sub’ (subject) field, which contains the user’s unique identifier that NiFi uses for policy assignment.

Parameters:

token_data (dict) – The full OAuth2 token response from service_login_oidc() when called with return_token_info=True

Returns:

The user identity UUID from the token’s ‘sub’ field

Return type:

str

Raises:

ValueError – If the token cannot be decoded or doesn’t contain expected fields

nipyapi.utils.filter_obj(obj, value, key, greedy=True)[source]

Implements a custom filter method because native datatypes don’t have consistently named or located fields.

Note that each object used by this function must be registered with identifier_types and identifiers in config

Parameters:

  • obj (varies) – the NiFi or NiFi-Registry object to filter on

  • value (str) – the String value to look for

  • key (str) – the object key to filter against

  • greedy (bool) – If True, the value will be matched anywhere in the string, if False it will require exact match

Returns: None if 0 matches, list if > 1, single Object entity if ==1

nipyapi.utils.format_timestamp(ts=None, fmt=None)[source]

Format a timestamp as a string.

Converts a timestamp to a formatted string. Defaults to ISO 8601 format with millisecond precision as expected by NiFi APIs.

Parameters:

  • ts (None or datetime or str) – Timestamp to format. None uses current UTC, datetime formats directly (assumes UTC if naive), str parses as ISO 8601.

  • fmt (str or None) – strftime format string. Defaults to ISO 8601 with milliseconds (“%Y-%m-%dT%H:%M:%S.{ms}Z”).

Returns:

Formatted timestamp string. Default format includes millisecond

precision and ‘Z’ suffix indicating UTC.

Return type:

str

Example:

>>> format_timestamp()  # Current time, ISO 8601
'2025-01-15T12:30:45.123Z'
>>> from datetime import datetime, timezone
>>> dt = datetime(2025, 1, 15, 12, 30, 45, 123456, tzinfo=timezone.utc)
>>> format_timestamp(dt)
'2025-01-15T12:30:45.123Z'
>>> format_timestamp(dt, fmt="%Y-%m-%d")
'2025-01-15'
>>> format_timestamp("2025-01-15T12:30:45.123Z", fmt="%Y-%m-%d")
'2025-01-15'
nipyapi.utils.fs_read(file_path)[source]

Convenience function to read an Object from a FilePath

Parameters:

file_path (str) – The Full path including filename to read from

Returns: The object that was read

nipyapi.utils.fs_write(obj, file_path, binary=False)[source]

Convenience function to write an Object to a FilePath.

Parameters:

  • obj (varies) – The Object to write out

  • file_path (str) – The Full path including filename to write to

  • binary (bool) – If True, write in binary mode (obj should be bytes). If False (default), write as text with encoding.

Returns:

The object that was written

Example:

# Write text content
nipyapi.utils.fs_write("hello world", "/tmp/test.txt")

# Write binary content
nipyapi.utils.fs_write(b"\x00\x01\x02", "/tmp/test.bin", binary=True)
nipyapi.utils.getenv(name: str, default: str | None = None) → str | None[source]

Enhanced environment variable getter with None handling.

Parameters:

  • name (str) – Environment variable name

  • default (Optional[str]) – Default value if variable not set

Returns:

Environment variable value or default

Return type:

Optional[str]

nipyapi.utils.getenv_bool(name: str, default: bool | None = None) → bool | None[source]

Parse environment variable as boolean.

Convenience wrapper around parse_bool() for environment variables. Handles common boolean environment variable patterns.

Parameters:

  • name (str) – Environment variable name

  • default (Optional[bool]) – Default value if variable not set

Returns:

Boolean value or default if not set

Return type:

Optional[bool]

Example:

>>> os.environ['MY_FLAG'] = '0'
>>> getenv_bool('MY_FLAG')  # False
>>> os.environ['MY_FLAG'] = 'true'
>>> getenv_bool('MY_FLAG')  # True
>>> getenv_bool('UNSET_FLAG', False)  # False
nipyapi.utils.infer_object_label_from_class(obj)[source]
Returns the expected STRING label for an object class required by certain

functions.

Parameters:

obj – The object to infer the name of

Returns:

str of the relevant name, or raises an AssertionError

nipyapi.utils.is_endpoint_up(endpoint_url)[source]

Tests if a URL is available for requests

A service is considered “up” if it responds with: - Success codes (200-399) - Authentication required codes (401, 403) - service is ready for auth - SSL certificate verification errors - service up but cert issues

SSL handshake failures (UNEXPECTED_EOF, etc.) indicate service not ready.

Parameters:

endpoint_url (str) – The URL to test

Returns (bool): True if service is ready for requests, False if not

nipyapi.utils.is_uuid(value)[source]

Check if a string looks like a UUID.

This validates the format only (8-4-4-4-12 hexadecimal pattern), not UUID validity per RFC 4122. It does not verify version or variant bits.

Parameters:

value (str) – The string to check

Returns:

True if the string matches UUID format (8-4-4-4-12 hex pattern)

Return type:

bool

nipyapi.utils.load(obj, dto=None)[source]

Loads a serialised object back into native datatypes, and optionally imports it back into the native NiFi DTO

Warning: Using this on objects not produced by this Package may have unintended results! While efforts have been made to ensure that unsafe loading is not possible, no stringent security testing has been completed.

Parameters:

  • obj (dict, list) – The serialised object to import

  • dto (Optional [tuple{str, str}]) – A Tuple describing the service and

  • constructed. (object that should be)

  • = (e.g. dto)

Returns: Either the loaded object in native Python datatypes, or the

constructed native datatype object

nipyapi.utils.parse_bool(value, default=None)[source]

Parse a value as boolean.

Handles string values from CLI parameters (where fire passes –flag=false as the string “false”) and environment variables. Also passes through actual bool values unchanged.

This function addresses the CLI boolean parameter gotcha where fire passes –flag=false as the string “false”, which is truthy in Python. Use this to safely convert CLI parameters to boolean.

Parameters:

  • value – The value to parse (str, bool, or None)

  • default – Value to return if input is None

Returns:

bool or default value if input is None

Example:

>>> parse_bool("false")  # False (handles CLI --flag=false)
>>> parse_bool("0")      # False
>>> parse_bool("yes")    # True
>>> parse_bool(True)     # True (passthrough)
>>> parse_bool(None, default=False)  # False
nipyapi.utils.resolve_entity(identifier, getter_func, expected_type, strict=True, greedy=True, identifier_type='auto')[source]

Resolve an identifier to an entity object.

This utility provides consistent entity resolution across nipyapi modules. Given an identifier (string or object) and a getter function, it returns the resolved entity object or raises an appropriate error.

Parameters:

  • identifier – Entity object, ID string, or name string.

  • getter_func – A getter function with signature (identifier, identifier_type, greedy) e.g., nipyapi.canvas.get_processor, nipyapi.versioning.get_registry_client

  • expected_type – The class the result should be (e.g., nipyapi.nifi.ProcessorEntity). Used to validate object inputs and for error messages.

  • strict (bool) – Error handling mode: - True (default): 0 matches -> ValueError, multiple -> ValueError - False: 0 matches -> None, multiple -> list

  • greedy (bool) – For name matching, True for partial match, False for exact.

  • identifier_type (str) – How to interpret string identifiers: - “auto” (default): UUID format -> ID lookup, else -> name lookup - “id”: Force ID lookup - “name”: Force name lookup

Returns:

Entity object of expected_type (or None/list if strict=False).

Raises:

  • TypeError – If identifier is not a string or instance of expected_type.

  • ValueError – If strict=True and 0 or multiple matches are found.

Examples

# In canvas.py action functions from nipyapi import utils, nifi processor = utils.resolve_entity(

processor, get_processor, nifi.ProcessorEntity, strict=True

)

# Entity objects pass through unchanged proc_entity = get_processor(“my-processor”) result = utils.resolve_entity(proc_entity, get_processor, nifi.ProcessorEntity) # result is proc_entity

nipyapi.utils.resolve_relative_paths(file_path, root_path=None)[source]

Convert a relative path to absolute, leave absolute paths unchanged.

Essential for SSL/TLS certificate configuration where libraries typically require absolute paths to avoid ambiguity about file locations.

Parameters:

  • file_path (str or None) – File path to resolve

  • root_path (str, optional) – Root directory for relative path resolution. Defaults to PROJECT_ROOT_DIR if not specified.

Returns:

Absolute path if input was a relative path string,

unchanged if input was absolute path or None.

Return type:

str or None

Example:

>>> resolve_relative_paths('certs/ca.pem', '/project')
'/project/certs/ca.pem'
>>> resolve_relative_paths('/etc/ssl/ca.pem')
'/etc/ssl/ca.pem'
>>> resolve_relative_paths(None)
None
nipyapi.utils.resolve_schedule_state(scheduled, true_state, false_state, valid_states=None)[source]

Normalize a scheduling parameter to a state string.

This utility provides consistent handling of the ‘scheduled’ parameter across nipyapi scheduling functions, accepting bool or string values.

Parameters:

  • scheduled – The scheduling input, either: - bool: True maps to true_state, False maps to false_state - str: Must be one of valid_states (case-insensitive)

  • true_state (str) – The state string when scheduled=True (e.g., “RUNNING”, “ENABLED”)

  • false_state (str) – The state string when scheduled=False (e.g., “STOPPED”, “DISABLED”)

  • valid_states (tuple, optional) – Tuple of valid state strings. If None, defaults to (true_state, false_state).

Returns:

The normalized state string (uppercase).

Return type:

str

Raises:

ValueError – If scheduled is not a bool or valid state string.

Examples

# Processor scheduling state = resolve_schedule_state(True, “RUNNING”, “STOPPED”,

(“RUNNING”, “STOPPED”, “DISABLED”, “RUN_ONCE”)) # -> “RUNNING”

state = resolve_schedule_state(“run_once”, “RUNNING”, “STOPPED”,

(“RUNNING”, “STOPPED”, “DISABLED”, “RUN_ONCE”)) # -> “RUN_ONCE”

# Controller scheduling state = resolve_schedule_state(False, “ENABLED”, “DISABLED”) # -> “DISABLED”

nipyapi.utils.set_endpoint(endpoint_url, ssl=False, login=False, username=None, password=None)[source]

Sets the endpoint when switching between instances of NiFi or other projects.

Parameters:

  • endpoint_url (str) – The URL to set as the endpoint

  • ssl (bool) – Whether to use SSL context for HTTPS connections

  • login (bool) – Whether to attempt token-based login

  • username (str) – The username to use for login

  • password (str) – The password to use for login

Returns (bool): True for success

nipyapi.utils.validate_parameters_versioning_support(verify_nifi=True, verify_registry=True)[source]

Convenience method to check if Parameters are supported :param verify_nifi: If True, check NiFi meets the min version :type verify_nifi: bool :param verify_registry: If True, check Registry meets the min version :type verify_registry: bool

nipyapi.utils.wait_to_complete(test_function, *args, **kwargs)[source]

Implements a basic return loop for a given function which is capable of a True|False output

Parameters:

  • test_function – Function which returns a bool once the target state is reached

  • delay (int) – The number of seconds between each attempt, defaults to config.short_retry_delay

  • max_wait (int) – the maximum number of seconds before issuing a Timeout, defaults to config.short_max_wait

  • *args – Any args to pass through to the test function

  • **kwargs – Any Keword Args to pass through to the test function

Returns (bool): True for success, False for not