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.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)[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

Returns: The object that was written

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 using JSON-style interpretation.

Handles common boolean environment variable patterns and uses json.loads() for the standard ‘true’/’false’ cases that most programmers understand.

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.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.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.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