Hey there, Anyone help me please with this issue?
Error while creating an InvokeScriptedProcessor using Nipyapi.
the Error throws when canvas. create() is called.
def create_processor(processor: ProcessorBase,
process_group: ProcessGroup) -> bool:
"""
Create a processor after adding the required properties to the processor configuration.
:param processor: ProcessorBase object with information about the nifi processor to be created.
:param process_group: ProcessGroup object such the processor will be assigned to it.
:return new_processor (ProcessorEntity): Nifi processor.
"""
try:
# Gets the abstract object describing a Processor or list thereof.
processor_type = canvas.get_processor_type(processor.name, identifier_type='name')
# Gets the first element of the list returned in case of list.
processor_type = processor_type[0] if type(processor_type) == list else processor_type
# Creates nifi processor.
new_processor = canvas.create_processor(process_group.process_group_entity,
processor_type,
processor.location,
config=processor.configuration)
processor.processor_entity = new_processor
return True
except Exception as ex:
print("Error during the creation of a processor ", str(ex))
return False
