Hi @Dan Chaffelson thanks for the quick response here is my code , this is just a POC so it may seems somewhat unstructured, just a playground
import os
import sys
import getpass
import json
import smtplib
from nipyapi import nifi, config, templates, canvas
from config import ConfigIni
# Disable urllib3 certificate warnings
from requests.packages.urllib3 import disable_warnings
disable_warnings()
class NifiInstance:
""" The NifiInstance class facilitating easy to use
methods utilizing the NiPyApi (https://github.com/Chaffelson/nipyapi)
wrapper library.
Arguments:
url (str): Nifi host url, defaults to environment variable `NIFI_HOST`.
username (str): Nifi username, defaults to environment variable `NIFI_USERNAME`.
password (str): Nifi password, defaults to environment variable `NIFI_PASSWORD`.
verify_ssl (bool): Whether to verify SSL connection - UNUSED as of now.
"""
def __init__(self, url=None, username=None, password=None, verify_ssl=False):
config.nifi_config.host = self._get_url(url)
config.nifi_config.verify_ssl = verify_ssl
config.nifi_config.username = username
self._authenticate(username, password)
def _get_url(self, url):
if not url:
try:
url = os.environ['NIFI_HOST']
except KeyError:
url = input('Nifi host: ')
if not '/nifi-api' in url:
if not url[-1] == '/':
url = url + '/'
url = url + 'nifi-api'
return url
def _authenticate(self, username=None, password=None):
if not username:
try:
config.nifi_config.username = os.environ['NIFI_USERNAME']
except KeyError:
config.nifi_config.username = input('Username: ')
if not password:
try:
password = os.environ['NIFI_PASSWORD']
except KeyError:
password = getpass.getpass('Password: ')
access_token = None
try:
access_token = nifi.AccessApi().create_access_token(username=config.nifi_config.username,password=password)
except nifi.rest.ApiException as e:
print('Exception when calling AccessApi->create_access_token: %s\n'.format(e))
config.nifi_config.api_key[username] = access_token
config.nifi_config.api_client = nifi.ApiClient(header_name='Authorization', header_value='Bearer {}'.format(access_token))
def list_processors_in_processorgroup(self,pg_id=None):
listen = canvas.list_all_processors(pg_id)
#jlisten= json.loads(listen)
for item in listen:
#print (str(item.id))
print (str(item.status.name))
def processor_status(self,p_id=None):
pro=canvas.get_processor(p_id, 'id')
return pro.status.run_status
#get urls and froups to list from ini file.
start_init = ConfigIni('c:/temp/nipyapi/monitor.ini')
#groups are returned as list of groups
groups= start_init.get_group_id()
url = start_init.get_url()
n = NifiInstance(url, 'myuser','mypassword')
#test list_all_processor
test=canvas.list_all_processors('5b641351-34d0-3def-a376-7824fbe9cc0f')
for item in test:
print (str(item .id))
my nipyapi version is C:\Temp\nipyapi (nipyapi) λ pip freeze | grep nipyapi nipyapi==0.11.0 C:\Temp\nipyapi
... View more
