You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2024/11/15 14:13:55 UTC
(nifi-minifi-cpp) 03/06: MINIFICPP-2279 Add support for RecordTransform NiFi python processors
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 153c16f22f18bb8592ae0abe66e4acf411be18d9
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Aug 21 17:00:13 2024 +0200
MINIFICPP-2279 Add support for RecordTransform NiFi python processors
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1863
---
docker/test/integration/cluster/ImageStore.py | 2 +
.../integration/features/fetch_modbus_tcp.feature | 3 +-
docker/test/integration/features/python.feature | 24 +++
docker/test/integration/features/steps/steps.py | 21 ++-
...onRecordSetWriter.py => JsonRecordSetReader.py} | 9 +-
.../minifi/controllers/JsonRecordSetWriter.py | 4 +-
.../SetRecordField.py} | 18 ++-
.../integration/resources/python/SetRecordField.py | 57 +++++++
extensions/python/PYTHON.md | 4 +-
extensions/python/PythonBindings.cpp | 6 +-
.../pythonprocessors/nifiapi/flowfilesource.py | 6 -
.../pythonprocessors/nifiapi/flowfiletransform.py | 6 -
.../pythonprocessors/nifiapi/recordtransform.py | 152 ++++++++++++++++++
extensions/python/types/PyProcessContext.cpp | 8 +
extensions/python/types/PyProcessContext.h | 2 +
extensions/python/types/PyProcessSession.cpp | 36 +++--
extensions/python/types/PyProcessSession.h | 3 +-
extensions/python/types/PyRecordSetReader.cpp | 112 +++++++++++++
.../{PyProcessContext.h => PyRecordSetReader.h} | 24 ++-
extensions/python/types/PyRecordSetWriter.cpp | 114 +++++++++++++
.../{PyProcessContext.h => PyRecordSetWriter.h} | 24 ++-
extensions/python/types/Types.h | 9 ++
.../tests/unit/RecordSetTests.cpp | 177 +++++++++++++++++++++
libminifi/include/core/Record.h | 22 ++-
libminifi/include/core/RecordField.h | 3 +
libminifi/src/core/RecordField.cpp | 99 ++++++++++++
26 files changed, 862 insertions(+), 83 deletions(-)
diff --git a/docker/test/integration/cluster/ImageStore.py b/docker/test/integration/cluster/ImageStore.py
index caca3e97d..781ce420d 100644
--- a/docker/test/integration/cluster/ImageStore.py
+++ b/docker/test/integration/cluster/ImageStore.py
@@ -171,6 +171,7 @@ class ImageStore:
COPY CreateNothing.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/CreateNothing.py
COPY FailureWithContent.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/FailureWithContent.py
COPY TransferToOriginal.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/TransferToOriginal.py
+ COPY SetRecordField.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/SetRecordField.py
RUN wget {parse_document_url} --directory-prefix=/opt/minifi/minifi-current/minifi-python/nifi_python_processors && \\
wget {chunk_document_url} --directory-prefix=/opt/minifi/minifi-current/minifi-python/nifi_python_processors && \\
echo 'langchain<=0.17.0' > /opt/minifi/minifi-current/minifi-python/nifi_python_processors/requirements.txt && \\
@@ -205,6 +206,7 @@ class ImageStore:
build_full_python_resource_path("CreateNothing.py"),
build_full_python_resource_path("FailureWithContent.py"),
build_full_python_resource_path("TransferToOriginal.py"),
+ build_full_python_resource_path("SetRecordField.py"),
])
def __build_http_proxy_image(self):
diff --git a/docker/test/integration/features/fetch_modbus_tcp.feature b/docker/test/integration/features/fetch_modbus_tcp.feature
index 0ecb56b10..99ce18f46 100644
--- a/docker/test/integration/features/fetch_modbus_tcp.feature
+++ b/docker/test/integration/features/fetch_modbus_tcp.feature
@@ -21,9 +21,10 @@ Feature: Minifi C++ can act as a modbus tcp master
Scenario: MiNiFi can fetch data from a modbus slave
Given a FetchModbusTcp processor
- And a JsonRecordSetWriter controller service is set up for FetchModbusTcp
+ And a JsonRecordSetWriter controller service is set up with "One Line Per Object" output grouping
And a PutFile processor with the "Directory" property set to "/tmp/output"
And the "Unit Identifier" property of the FetchModbusTcp processor is set to "255"
+ And the "Record Set Writer" property of the FetchModbusTcp processor is set to "JsonRecordSetWriter"
And there is an accessible PLC with modbus enabled
And PLC register has been set with h@52=123 command
And PLC register has been set with h@5678/f=1.75 command
diff --git a/docker/test/integration/features/python.feature b/docker/test/integration/features/python.feature
index ab73628dd..5c63a4e31 100644
--- a/docker/test/integration/features/python.feature
+++ b/docker/test/integration/features/python.feature
@@ -222,3 +222,27 @@ Feature: MiNiFi can use python processors in its flows
When all instances start up
Then the Minifi logs contain the following message: "Result relationship cannot be 'original', it is reserved for the original flow file, and transferred automatically in non-failure cases." in less than 60 seconds
+
+ @USE_NIFI_PYTHON_PROCESSORS
+ Scenario: MiNiFi C++ supports RecordTransform native python processors
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"group": "group1", "name": "John"}\n{"group": "group1", "name": "Jane"}\n{"group": "group2", "name": "Kyle"}\n{"name": "Zoe"}' is present in '/tmp/input'
+ And a file with the content '{"group": "group1", "name": "Steve"}\n{}' is present in '/tmp/input'
+ And a SetRecordField processor with the "Record Reader" property set to "JsonRecordSetReader"
+ And the "Record Writer" property of the SetRecordField processor is set to "JsonRecordSetWriter"
+ And a JsonRecordSetReader controller service is set up
+ And a JsonRecordSetWriter controller service is set up with "Array" output grouping
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And the "Log Payload" property of the LogAttribute processor is set to "true"
+ And python is installed on the MiNiFi agent with a pre-created virtualenv
+
+ And the "success" relationship of the GetFile processor is connected to the SetRecordField
+ And the "success" relationship of the SetRecordField processor is connected to the LogAttribute
+
+ When all instances start up
+
+ Then the Minifi logs contain the following message: '[{"group":"group1","name":"John"},{"group":"group1","name":"Jane"}]' in less than 60 seconds
+ And the Minifi logs contain the following message: '[{"group":"group2","name":"Kyle"}]' in less than 5 seconds
+ And the Minifi logs contain the following message: '[{"name":"Zoe"}]' in less than 5 seconds
+ And the Minifi logs contain the following message: '[{"group":"group1","name":"Steve"}]' in less than 5 seconds
+ And the Minifi logs contain the following message: '[{}]' in less than 5 seconds
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 92dd64548..a56afa52a 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -25,6 +25,7 @@ from minifi.controllers.ElasticsearchCredentialsService import ElasticsearchCred
from minifi.controllers.ODBCService import ODBCService
from minifi.controllers.KubernetesControllerService import KubernetesControllerService
from minifi.controllers.JsonRecordSetWriter import JsonRecordSetWriter
+from minifi.controllers.JsonRecordSetReader import JsonRecordSetReader
from behave import given, then, when
from behave.model_describe import ModelDescriptor
@@ -271,6 +272,7 @@ def step_impl(context):
@given("a file with the content \"{content}\" is present in \"{path}\"")
+@given("a file with the content '{content}' is present in '{path}'")
@then("a file with the content \"{content}\" is placed in \"{path}\"")
def step_impl(context, content, path):
context.test.add_test_data(path, content)
@@ -401,14 +403,19 @@ def step_impl(context, processor_name):
processor.set_property('SSL Context Service', ssl_context_service.name)
-# RecordSetWriters
-@given("a JsonRecordSetWriter controller service is set up for {processor_name}")
-def step_impl(context, processor_name):
- json_record_set_writer = JsonRecordSetWriter()
+# Record set reader and writer
+@given("a JsonRecordSetWriter controller service is set up with \"{}\" output grouping")
+def step_impl(context, output_grouping: str):
+ json_record_set_writer = JsonRecordSetWriter(name="JsonRecordSetWriter", output_grouping=output_grouping)
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(json_record_set_writer)
- processor = context.test.get_node_by_name(processor_name)
- processor.controller_services.append(json_record_set_writer)
- processor.set_property('Record Set Writer', json_record_set_writer.name)
+
+@given("a JsonRecordSetReader controller service is set up")
+def step_impl(context):
+ json_record_set_reader = JsonRecordSetReader("JsonRecordSetReader")
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(json_record_set_reader)
# Kubernetes
diff --git a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py b/docker/test/integration/minifi/controllers/JsonRecordSetReader.py
similarity index 71%
copy from docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
copy to docker/test/integration/minifi/controllers/JsonRecordSetReader.py
index 9708c6cb6..909bf0e7c 100644
--- a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
+++ b/docker/test/integration/minifi/controllers/JsonRecordSetReader.py
@@ -17,8 +17,7 @@
from ..core.ControllerService import ControllerService
-class JsonRecordSetWriter(ControllerService):
- def __init__(self, name=None, cert=None, key=None, ca_cert=None, passphrase=None, use_system_cert_store=None):
- super(JsonRecordSetWriter, self).__init__(name=name)
- self.service_class = 'JsonRecordSetWriter'
- self.properties['Output Grouping'] = 'One Line Per Object'
+class JsonRecordSetReader(ControllerService):
+ def __init__(self, name=None):
+ super(JsonRecordSetReader, self).__init__(name=name)
+ self.service_class = 'JsonRecordSetReader'
diff --git a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py b/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
index 9708c6cb6..ca00764b5 100644
--- a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
+++ b/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
@@ -18,7 +18,7 @@ from ..core.ControllerService import ControllerService
class JsonRecordSetWriter(ControllerService):
- def __init__(self, name=None, cert=None, key=None, ca_cert=None, passphrase=None, use_system_cert_store=None):
+ def __init__(self, name=None, output_grouping='One Line Per Object'):
super(JsonRecordSetWriter, self).__init__(name=name)
self.service_class = 'JsonRecordSetWriter'
- self.properties['Output Grouping'] = 'One Line Per Object'
+ self.properties['Output Grouping'] = output_grouping
diff --git a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py b/docker/test/integration/minifi/processors/SetRecordField.py
similarity index 64%
copy from docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
copy to docker/test/integration/minifi/processors/SetRecordField.py
index 9708c6cb6..71875b2a4 100644
--- a/docker/test/integration/minifi/controllers/JsonRecordSetWriter.py
+++ b/docker/test/integration/minifi/processors/SetRecordField.py
@@ -12,13 +12,15 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+from ..core.Processor import Processor
-from ..core.ControllerService import ControllerService
-
-
-class JsonRecordSetWriter(ControllerService):
- def __init__(self, name=None, cert=None, key=None, ca_cert=None, passphrase=None, use_system_cert_store=None):
- super(JsonRecordSetWriter, self).__init__(name=name)
- self.service_class = 'JsonRecordSetWriter'
- self.properties['Output Grouping'] = 'One Line Per Object'
+class SetRecordField(Processor):
+ def __init__(self, context):
+ super(SetRecordField, self).__init__(
+ context=context,
+ clazz='SetRecordField',
+ class_prefix='org.apache.nifi.minifi.processors.nifi_python_processors.',
+ properties={},
+ schedule={'scheduling strategy': 'EVENT_DRIVEN'},
+ auto_terminate=[])
diff --git a/docker/test/integration/resources/python/SetRecordField.py b/docker/test/integration/resources/python/SetRecordField.py
new file mode 100644
index 000000000..dbb3a2ae6
--- /dev/null
+++ b/docker/test/integration/resources/python/SetRecordField.py
@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from nifiapi.properties import PropertyDescriptor
+from nifiapi.properties import StandardValidators
+from nifiapi.properties import ExpressionLanguageScope
+from nifiapi.recordtransform import RecordTransformResult
+from nifiapi.recordtransform import RecordTransform
+
+
+class SetRecordField(RecordTransform):
+ class Java:
+ implements = ['org.apache.nifi.python.processor.RecordTransform']
+
+ class ProcessorDetails:
+ version = '0.0.1-SNAPSHOT'
+
+ def __init__(self, **kwargs):
+ super().__init__()
+
+ def transform(self, context, record, schema, attributemap):
+ # Update dictionary based on the dynamic properties provided by user
+ for key in context.getProperties().keys():
+ if not key.dynamic:
+ continue
+
+ propname = key.name
+ record[propname] = context.getProperty(propname).evaluateAttributeExpressions(attributemap).getValue()
+
+ # Determine the partition
+ if 'group' in record:
+ partition = {'group': record['group']}
+ else:
+ partition = None
+
+ # Return the result
+ return RecordTransformResult(record=record, relationship='success', partition=partition)
+
+ def getDynamicPropertyDescriptor(self, name):
+ return PropertyDescriptor(
+ name=name,
+ description="Specifies the value to set for the '" + name + "' field",
+ expression_language_scope=ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ validators=[StandardValidators.ALWAYS_VALID]
+ )
diff --git a/extensions/python/PYTHON.md b/extensions/python/PYTHON.md
index 2b0f7c076..eb0289bfe 100644
--- a/extensions/python/PYTHON.md
+++ b/extensions/python/PYTHON.md
@@ -155,14 +155,14 @@ The SentimentAnalysis processor will perform a Vader Sentiment Analysis. This re
## Using NiFi Python Processors
-MiNiFi C++ supports the use of NiFi Python processors, that are inherited from the FlowFileTransform or the FlowFileSource base class. To use these processors, copy the Python processor module to the nifi_python_processors subdirectory of the python directory. By default, the python directory is ${minifi_root}/minifi-python. To see how to write NiFi Python processors, please refer to the Python Developer Guide under the [Apache NiFi documentation](https://nifi.apache.org/documentation/v2/).
+MiNiFi C++ supports the use of NiFi Python processors, that are inherited from the FlowFileTransform, RecordTransform or the FlowFileSource base class. To use these processors, copy the Python processor module to the nifi_python_processors subdirectory of the python directory. By default, the python directory is ${minifi_root}/minifi-python. To see how to write NiFi Python processors, please refer to the Python Developer Guide under the [Apache NiFi documentation](https://nifi.apache.org [...]
In the flow configuration these Python processors can be referenced by their fully qualified class name, which looks like this: org.apache.nifi.minifi.processors.nifi_python_processors.<package_name>.<processor_name>. For example, the fully qualified class name of the PromptChatGPT processor implemented in the file nifi_python_processors/PromptChatGPT.py is org.apache.nifi.minifi.processors.nifi_python_processors.PromptChatGPT. If a processor is copied under a subdirectory, because it is [...]
**NOTE:** The name of the NiFi Python processor file should match the class name in the file, otherwise the processor will not be found.
Due to some differences between the NiFi and MiNiFi C++ processors and implementation, there are some limitations using the NiFi Python processors:
-- Record based processors are not yet supported in MiNiFi C++, so the NiFi Python processors inherited from RecordTransform are not supported.
+- Schemas are not supported in MiNiFi C++, so the schemas are ignored and passed as None in the `transform` method in the RecordTransform NiFi Python processors.
- There are some validators in NiFi that are not present in MiNiFi C++, so some property validations will be missing using the NiFi Python processors.
- MiNiFi C++ only supports expression language with flow file attributes, so only FLOWFILE_ATTRIBUTES expression language scope is supported, otherwise the expression language will not be evaluated.
- MiNiFi C++ does not support property dependencies, so the property dependencies will be ignored. If a property depends on another property, the property will not be required.
diff --git a/extensions/python/PythonBindings.cpp b/extensions/python/PythonBindings.cpp
index 6cd3d069d..873188faf 100644
--- a/extensions/python/PythonBindings.cpp
+++ b/extensions/python/PythonBindings.cpp
@@ -28,6 +28,8 @@
#include "types/PyStateManager.h"
#include "types/PyDataConverter.h"
#include "types/PySSLContextService.h"
+#include "types/PyRecordSetReader.h"
+#include "types/PyRecordSetWriter.h"
namespace org::apache::nifi::minifi::extensions::python {
extern "C" {
@@ -62,7 +64,9 @@ PyInit_minifi_native(void) {
std::make_pair(PyInputStream::typeObject(), "InputStream"),
std::make_pair(PyOutputStream::typeObject(), "OutputStream"),
std::make_pair(PyStateManager::typeObject(), "StateManager"),
- std::make_pair(PySSLContextService::typeObject(), "SSLContextService")
+ std::make_pair(PySSLContextService::typeObject(), "SSLContextService"),
+ std::make_pair(PyRecordSetReader::typeObject(), "RecordSetReader"),
+ std::make_pair(PyRecordSetWriter::typeObject(), "RecordSetWriter")
});
for (const auto& type : types) {
diff --git a/extensions/python/pythonprocessors/nifiapi/flowfilesource.py b/extensions/python/pythonprocessors/nifiapi/flowfilesource.py
index be676d9ff..74041d780 100644
--- a/extensions/python/pythonprocessors/nifiapi/flowfilesource.py
+++ b/extensions/python/pythonprocessors/nifiapi/flowfilesource.py
@@ -40,12 +40,6 @@ class FlowFileSourceResult:
class FlowFileSource(ProcessorBase):
- # These will be added through the python bindings using C API
- logger = None
- REL_SUCCESS = None
- REL_FAILURE = None
- REL_ORIGINAL = None
-
def onTrigger(self, context: ProcessContext, session: ProcessSession):
context_proxy = ProcessContextProxy(context, self)
try:
diff --git a/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py b/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
index be17a266e..c72ebfb54 100644
--- a/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
+++ b/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
@@ -41,12 +41,6 @@ class FlowFileTransformResult:
class FlowFileTransform(ProcessorBase):
- # These will be added through the python bindings using C API
- logger = None
- REL_SUCCESS = None
- REL_FAILURE = None
- REL_ORIGINAL = None
-
def onTrigger(self, context: ProcessContext, session: ProcessSession):
original_flow_file = session.get()
if not original_flow_file:
diff --git a/extensions/python/pythonprocessors/nifiapi/recordtransform.py b/extensions/python/pythonprocessors/nifiapi/recordtransform.py
new file mode 100644
index 000000000..bdeb2ffb6
--- /dev/null
+++ b/extensions/python/pythonprocessors/nifiapi/recordtransform.py
@@ -0,0 +1,152 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import traceback
+import json
+from abc import abstractmethod
+from minifi_native import ProcessContext, ProcessSession, Processor
+from .processorbase import ProcessorBase
+from .properties import FlowFile as FlowFileProxy
+from .properties import ProcessContext as ProcessContextProxy
+from .properties import PropertyDescriptor
+
+
+class __RecordTransformResult__:
+ def __init__(self, processor_result, recordJson):
+ self.processor_result = processor_result
+ self.recordJson = recordJson
+
+ def getRecordJson(self):
+ return self.recordJson
+
+ def getSchema(self):
+ return self.processor_result.schema
+
+ def getRelationship(self):
+ return self.processor_result.relationship
+
+ def getPartition(self):
+ return self.processor_result.partition
+
+
+class RecordTransformResult:
+ def __init__(self, record=None, schema=None, relationship="success", partition=None):
+ self.record = record
+ self.schema = schema
+ self.relationship = relationship
+ self.partition = partition
+
+ def getRecord(self):
+ return self.record
+
+ def getSchema(self):
+ return self.schema
+
+ def getRelationship(self):
+ return self.relationship
+
+ def getPartition(self):
+ return self.partition
+
+
+class RecordTransform(ProcessorBase):
+ RECORD_READER = PropertyDescriptor(
+ name='Record Reader',
+ display_name='Record Reader',
+ description='''Specifies the Controller Service to use for reading incoming data''',
+ required=True,
+ controller_service_definition='RecordSetReader'
+ )
+ RECORD_WRITER = PropertyDescriptor(
+ name='Record Writer',
+ display_name='Record Writer',
+ description='''Specifies the Controller Service to use for writing out the records''',
+ required=True,
+ controller_service_definition='RecordSetWriter',
+ )
+
+ def onInitialize(self, processor: Processor):
+ super(RecordTransform, self).onInitialize(processor)
+ processor.addProperty(self.RECORD_READER.name, self.RECORD_READER.description, None, self.RECORD_READER.required, False, False, None, None, self.RECORD_READER.controllerServiceDefinition)
+ processor.addProperty(self.RECORD_WRITER.name, self.RECORD_WRITER.description, None, self.RECORD_WRITER.required, False, False, None, None, self.RECORD_WRITER.controllerServiceDefinition)
+
+ def onTrigger(self, context: ProcessContext, session: ProcessSession):
+ flow_file = session.get()
+ if not flow_file:
+ return
+
+ context_proxy = ProcessContextProxy(context, self)
+ record_reader = context_proxy.getProperty(self.RECORD_READER).asControllerService()
+ if not record_reader:
+ self.logger.error("Record Reader property is invalid")
+ session.transfer(flow_file, self.REL_FAILURE)
+ return
+ record_writer = context_proxy.getProperty(self.RECORD_WRITER).asControllerService()
+ if not record_writer:
+ self.logger.error("Record Writer property is invalid")
+ session.transfer(flow_file, self.REL_FAILURE)
+ return
+
+ try:
+ record_list = record_reader.read(flow_file, session)
+ if record_list is None:
+ self.logger.error("Reading flow file records returned None")
+ session.transfer(flow_file, self.REL_FAILURE)
+ return
+ except Exception:
+ self.logger.error("Failed to read flow file records due to the following error:\n{}".format(traceback.format_exc()))
+ session.transfer(flow_file, self.REL_FAILURE)
+ return
+
+ flow_file_proxy = FlowFileProxy(session, flow_file)
+ results = []
+ for record in record_list:
+ record_json = json.loads(record)
+ try:
+ result = self.transform(context_proxy, record_json, None, flow_file_proxy)
+ result_record = result.getRecord()
+ resultjson = None if result_record is None else json.dumps(result_record)
+ results.append(__RecordTransformResult__(result, resultjson))
+ except Exception:
+ self.logger.error("Failed to transform record due to the following error:\n{}".format(traceback.format_exc()))
+ session.transfer(flow_file, self.REL_FAILURE)
+ return
+
+ partitions = []
+ partitioned_results_list = []
+ for result in results:
+ if result.getRecordJson() is None:
+ continue
+ record_partition = result.getPartition()
+ if record_partition in partitions:
+ partition_index = partitions.index(record_partition)
+ partitioned_results_list[partition_index].append(result)
+ else:
+ partitions.append(record_partition)
+ partitioned_results_list.append([result])
+
+ for single_partition_results in partitioned_results_list:
+ partitioned_flow_file = session.create(flow_file)
+ record_writer.write([result.getRecordJson() for result in single_partition_results], partitioned_flow_file, session)
+ if result.getRelationship() == "success":
+ session.transfer(partitioned_flow_file, self.REL_SUCCESS)
+ else:
+ session.transferToCustomRelationship(partitioned_flow_file, result.getRelationship())
+
+ session.transfer(flow_file, self.REL_ORIGINAL)
+
+ @abstractmethod
+ def transform(self, context: ProcessContextProxy, record_json, schema, flowFile: FlowFileProxy) -> RecordTransformResult:
+ pass
diff --git a/extensions/python/types/PyProcessContext.cpp b/extensions/python/types/PyProcessContext.cpp
index aa22fb39f..f2ccbaf7f 100644
--- a/extensions/python/types/PyProcessContext.cpp
+++ b/extensions/python/types/PyProcessContext.cpp
@@ -22,6 +22,8 @@ a * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#include "PyStateManager.h"
#include "PyScriptFlowFile.h"
#include "core/Processor.h"
+#include "controllers/RecordSetReader.h"
+#include "controllers/RecordSetWriter.h"
extern "C" {
namespace org::apache::nifi::minifi::extensions::python {
@@ -128,6 +130,12 @@ PyObject* PyProcessContext::getControllerService(PyProcessContext* self, PyObjec
if (controller_service_type_str == "SSLContextService") {
auto ssl_ctx_service = std::dynamic_pointer_cast<controllers::SSLContextService>(controller_service);
return object::returnReference(std::weak_ptr(ssl_ctx_service));
+ } else if (controller_service_type_str == "RecordSetReader") {
+ auto record_set_reader = std::dynamic_pointer_cast<core::RecordSetReader>(controller_service);
+ return object::returnReference(std::weak_ptr(record_set_reader));
+ } else if (controller_service_type_str == "RecordSetWriter") {
+ auto record_set_writer = std::dynamic_pointer_cast<core::RecordSetWriter>(controller_service);
+ return object::returnReference(std::weak_ptr(record_set_writer));
}
}
diff --git a/extensions/python/types/PyProcessContext.h b/extensions/python/types/PyProcessContext.h
index 8e3f515b4..faed8c021 100644
--- a/extensions/python/types/PyProcessContext.h
+++ b/extensions/python/types/PyProcessContext.h
@@ -20,6 +20,8 @@
#include "core/ProcessContext.h"
#include "PySSLContextService.h"
+#include "PyRecordSetReader.h"
+#include "PyRecordSetWriter.h"
#include "../PythonBindings.h"
namespace org::apache::nifi::minifi::extensions::python {
diff --git a/extensions/python/types/PyProcessSession.cpp b/extensions/python/types/PyProcessSession.cpp
index 00b9bf8a5..f2278b75d 100644
--- a/extensions/python/types/PyProcessSession.cpp
+++ b/extensions/python/types/PyProcessSession.cpp
@@ -30,11 +30,11 @@ namespace org::apache::nifi::minifi::extensions::python {
namespace core = org::apache::nifi::minifi::core;
PyProcessSession::PyProcessSession(core::ProcessSession& session)
- : session_(gsl::make_not_null(&session)) {
+ : session_(session) {
}
std::shared_ptr<core::FlowFile> PyProcessSession::get() {
- auto flow_file = session_->get();
+ auto flow_file = session_.get();
if (flow_file == nullptr) {
return nullptr;
@@ -51,7 +51,7 @@ void PyProcessSession::transfer(const std::shared_ptr<core::FlowFile>& flow_file
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->transfer(flow_file, relationship);
+ session_.transfer(flow_file, relationship);
}
void PyProcessSession::transferToCustomRelationship(const std::shared_ptr<core::FlowFile>& flow_file, const std::string& relationship_name) {
@@ -59,7 +59,7 @@ void PyProcessSession::transferToCustomRelationship(const std::shared_ptr<core::
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->transferToCustomRelationship(flow_file, relationship_name);
+ session_.transferToCustomRelationship(flow_file, relationship_name);
}
void PyProcessSession::read(const std::shared_ptr<core::FlowFile>& flow_file, BorrowedObject input_stream_callback) {
@@ -67,7 +67,7 @@ void PyProcessSession::read(const std::shared_ptr<core::FlowFile>& flow_file, Bo
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->read(flow_file, [&input_stream_callback](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
+ session_.read(flow_file, [&input_stream_callback](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
return Long(Callable(input_stream_callback.getAttribute("process"))(std::weak_ptr(input_stream))).asInt64();
});
}
@@ -77,13 +77,13 @@ void PyProcessSession::write(const std::shared_ptr<core::FlowFile>& flow_file, B
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->write(flow_file, [&output_stream_callback](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
+ session_.write(flow_file, [&output_stream_callback](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
return Long(Callable(output_stream_callback.getAttribute("process"))(std::weak_ptr(output_stream))).asInt64();
});
}
std::shared_ptr<core::FlowFile> PyProcessSession::create(const std::shared_ptr<core::FlowFile>& flow_file) {
- auto result = session_->create(flow_file.get());
+ auto result = session_.create(flow_file.get());
flow_files_.push_back(result);
return result;
@@ -94,14 +94,14 @@ std::shared_ptr<core::FlowFile> PyProcessSession::clone(const std::shared_ptr<co
throw std::runtime_error("Flow file to clone is nullptr");
}
- auto result = session_->clone(*flow_file);
+ auto result = session_.clone(*flow_file);
flow_files_.push_back(result);
return result;
}
void PyProcessSession::remove(const std::shared_ptr<core::FlowFile>& flow_file) {
- session_->remove(flow_file);
+ session_.remove(flow_file);
flow_files_.erase(ranges::remove_if(flow_files_, [&flow_file](const auto& ff)-> bool { return ff == flow_file; }), flow_files_.end());
}
@@ -111,7 +111,7 @@ std::string PyProcessSession::getContentsAsString(const std::shared_ptr<core::Fl
}
std::string content;
- session_->read(flow_file, [&content](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
+ session_.read(flow_file, [&content](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
content.resize(input_stream->size());
return gsl::narrow<int64_t>(input_stream->read(as_writable_bytes(std::span(content))));
});
@@ -119,7 +119,7 @@ std::string PyProcessSession::getContentsAsString(const std::shared_ptr<core::Fl
}
void PyProcessSession::putAttribute(const std::shared_ptr<core::FlowFile>& flow_file, std::string_view key, const std::string& value) {
- session_->putAttribute(*flow_file, key, value);
+ session_.putAttribute(*flow_file, key, value);
}
extern "C" {
@@ -178,14 +178,24 @@ PyObject* PyProcessSessionObject::get(PyProcessSessionObject* self, PyObject*) {
return object::returnReference(nullptr);
}
-PyObject* PyProcessSessionObject::create(PyProcessSessionObject* self, PyObject*) {
+PyObject* PyProcessSessionObject::create(PyProcessSessionObject* self, PyObject* args) {
auto session = self->process_session_.lock();
if (!session) {
PyErr_SetString(PyExc_AttributeError, "tried reading process session outside 'on_trigger'");
return nullptr;
}
- if (auto flow_file = session->create(nullptr))
+ std::shared_ptr<core::FlowFile> parent_flow_file;
+ auto arg_size = PyTuple_Size(args);
+ if (arg_size > 0) {
+ PyObject* script_flow_file = nullptr;
+ if (!PyArg_ParseTuple(args, "O!", PyScriptFlowFile::typeObject(), &script_flow_file)) {
+ return nullptr;
+ }
+ parent_flow_file = reinterpret_cast<PyScriptFlowFile*>(script_flow_file)->script_flow_file_.lock();
+ }
+
+ if (auto flow_file = session->create(parent_flow_file))
return object::returnReference(std::weak_ptr(flow_file));
return object::returnReference(nullptr);
}
diff --git a/extensions/python/types/PyProcessSession.h b/extensions/python/types/PyProcessSession.h
index 1a8896679..c2f0bbe9d 100644
--- a/extensions/python/types/PyProcessSession.h
+++ b/extensions/python/types/PyProcessSession.h
@@ -39,10 +39,11 @@ class PyProcessSession {
void remove(const std::shared_ptr<core::FlowFile>& flow_file);
std::string getContentsAsString(const std::shared_ptr<core::FlowFile>& flow_file);
void putAttribute(const std::shared_ptr<core::FlowFile>& flow_file, std::string_view key, const std::string& value);
+ core::ProcessSession& getSession() const { return session_; }
private:
std::vector<std::shared_ptr<core::FlowFile>> flow_files_;
- gsl::not_null<core::ProcessSession*> session_;
+ core::ProcessSession& session_;
};
struct PyProcessSessionObject {
diff --git a/extensions/python/types/PyRecordSetReader.cpp b/extensions/python/types/PyRecordSetReader.cpp
new file mode 100644
index 000000000..d36f6a11f
--- /dev/null
+++ b/extensions/python/types/PyRecordSetReader.cpp
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+a * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "PyRecordSetReader.h"
+#include "rapidjson/writer.h"
+#include "rapidjson/stream.h"
+
+#include "PyProcessSession.h"
+#include "PyScriptFlowFile.h"
+
+extern "C" {
+namespace org::apache::nifi::minifi::extensions::python {
+
+static PyMethodDef PyRecordSetReader_methods[] = { // NOLINT(cppcoreguidelines-avoid-c-arrays)
+ {"read", (PyCFunction) PyRecordSetReader::read, METH_VARARGS, nullptr},
+ {} /* Sentinel */
+};
+
+static PyType_Slot PyRecordSetReaderTypeSpecSlots[] = { // NOLINT(cppcoreguidelines-avoid-c-arrays)
+ {Py_tp_dealloc, reinterpret_cast<void*>(pythonAllocatedInstanceDealloc<PyRecordSetReader>)},
+ {Py_tp_init, reinterpret_cast<void*>(PyRecordSetReader::init)},
+ {Py_tp_methods, reinterpret_cast<void*>(PyRecordSetReader_methods)},
+ {Py_tp_new, reinterpret_cast<void*>(newPythonAllocatedInstance<PyRecordSetReader>)},
+ {} /* Sentinel */
+};
+
+static PyType_Spec PyRecordSetReaderTypeSpec{
+ .name = "minifi_native.RecordSetReader",
+ .basicsize = sizeof(PyRecordSetReader),
+ .itemsize = 0,
+ .flags = Py_TPFLAGS_DEFAULT,
+ .slots = PyRecordSetReaderTypeSpecSlots
+};
+
+int PyRecordSetReader::init(PyRecordSetReader* self, PyObject* args, PyObject*) {
+ gsl_Expects(self && args);
+ PyObject* weak_ptr_capsule = nullptr;
+ if (!PyArg_ParseTuple(args, "O", &weak_ptr_capsule)) {
+ return -1;
+ }
+
+ auto record_set_reader = PyCapsule_GetPointer(weak_ptr_capsule, HeldTypeName);
+ if (!record_set_reader)
+ return -1;
+ self->record_set_reader_ = *static_cast<HeldType*>(record_set_reader);
+ return 0;
+}
+
+PyObject* PyRecordSetReader::read(PyRecordSetReader* self, PyObject* args) {
+ gsl_Expects(self && args);
+ auto record_set_reader = self->record_set_reader_.lock();
+ if (!record_set_reader) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading record set reader outside 'on_trigger'");
+ return nullptr;
+ }
+
+ PyObject* script_flow_file = nullptr;
+ PyObject* py_session = nullptr;
+ if (!PyArg_ParseTuple(args, "O!O!", PyScriptFlowFile::typeObject(), &script_flow_file, PyProcessSessionObject::typeObject(), &py_session)) {
+ return nullptr;
+ }
+
+ const auto flow_file = reinterpret_cast<PyScriptFlowFile*>(script_flow_file)->script_flow_file_.lock();
+ if (!flow_file) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading FlowFile outside 'on_trigger'");
+ return nullptr;
+ }
+
+ const auto process_session = reinterpret_cast<PyProcessSessionObject*>(py_session)->process_session_.lock();
+ if (!process_session) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading ProcessSession outside 'on_trigger'");
+ return nullptr;
+ }
+
+ auto read_result = record_set_reader->read(flow_file, process_session->getSession());
+ if (!read_result) {
+ std::string error_message = "failed to read record set with the following error: " + read_result.error().message();
+ PyErr_SetString(PyExc_RuntimeError, error_message.c_str());
+ return nullptr;
+ }
+
+ auto records = OwnedList::create();
+ for (const auto& record : read_result.value()) {
+ rapidjson::StringBuffer buffer;
+ rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+ record.toJson().Accept(writer);
+ records.append(std::string{buffer.GetString(), buffer.GetSize()});
+ }
+ return object::returnReference(records);
+}
+
+PyTypeObject* PyRecordSetReader::typeObject() {
+ static OwnedObject PyRecordSetReaderType{PyType_FromSpec(&PyRecordSetReaderTypeSpec)};
+ return reinterpret_cast<PyTypeObject*>(PyRecordSetReaderType.get());
+}
+
+} // namespace org::apache::nifi::minifi::extensions::python
+} // extern "C"
diff --git a/extensions/python/types/PyProcessContext.h b/extensions/python/types/PyRecordSetReader.h
similarity index 55%
copy from extensions/python/types/PyProcessContext.h
copy to extensions/python/types/PyRecordSetReader.h
index 8e3f515b4..b0488ff5e 100644
--- a/extensions/python/types/PyProcessContext.h
+++ b/extensions/python/types/PyRecordSetReader.h
@@ -18,34 +18,28 @@
#include <memory>
-#include "core/ProcessContext.h"
-#include "PySSLContextService.h"
+#include "controllers/RecordSetReader.h"
#include "../PythonBindings.h"
namespace org::apache::nifi::minifi::extensions::python {
-struct PyProcessContext {
- PyProcessContext() {}
- using HeldType = core::ProcessContext*;
- static constexpr const char* HeldTypeName = "PyProcessContext::HeldType";
+struct PyRecordSetReader {
+ PyRecordSetReader() {}
+ using HeldType = std::weak_ptr<core::RecordSetReader>;
+ static constexpr const char* HeldTypeName = "PyRecordSetReader::HeldType";
PyObject_HEAD
- HeldType process_context_;
+ HeldType record_set_reader_;
- static int init(PyProcessContext* self, PyObject* args, PyObject* kwds);
+ static int init(PyRecordSetReader* self, PyObject* args, PyObject* kwds);
- static PyObject* getProperty(PyProcessContext* self, PyObject* args);
- static PyObject* getStateManager(PyProcessContext* self, PyObject* args);
- static PyObject* getControllerService(PyProcessContext* self, PyObject* args);
- static PyObject* getName(PyProcessContext* self, PyObject* args);
- static PyObject* getProperties(PyProcessContext* self, PyObject* args);
- static PyObject* yieldResources(PyProcessContext* self, PyObject* args);
+ static PyObject* read(PyRecordSetReader* self, PyObject* args);
static PyTypeObject* typeObject();
};
namespace object {
template<>
-struct Converter<PyProcessContext::HeldType> : public HolderTypeConverter<PyProcessContext> {};
+struct Converter<PyRecordSetReader::HeldType> : public HolderTypeConverter<PyRecordSetReader> {};
} // namespace object
} // namespace org::apache::nifi::minifi::extensions::python
diff --git a/extensions/python/types/PyRecordSetWriter.cpp b/extensions/python/types/PyRecordSetWriter.cpp
new file mode 100644
index 000000000..ffd784e7b
--- /dev/null
+++ b/extensions/python/types/PyRecordSetWriter.cpp
@@ -0,0 +1,114 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+a * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "PyRecordSetWriter.h"
+
+#include "PyScriptFlowFile.h"
+#include "PyProcessSession.h"
+#include "core/Record.h"
+#include "rapidjson/document.h"
+
+extern "C" {
+namespace org::apache::nifi::minifi::extensions::python {
+
+static PyMethodDef PyRecordSetWriter_methods[] = { // NOLINT(cppcoreguidelines-avoid-c-arrays)
+ {"write", (PyCFunction) PyRecordSetWriter::write, METH_VARARGS, nullptr},
+ {} /* Sentinel */
+};
+
+static PyType_Slot PyRecordSetWriterTypeSpecSlots[] = { // NOLINT(cppcoreguidelines-avoid-c-arrays)
+ {Py_tp_dealloc, reinterpret_cast<void*>(pythonAllocatedInstanceDealloc<PyRecordSetWriter>)},
+ {Py_tp_init, reinterpret_cast<void*>(PyRecordSetWriter::init)},
+ {Py_tp_methods, reinterpret_cast<void*>(PyRecordSetWriter_methods)},
+ {Py_tp_new, reinterpret_cast<void*>(newPythonAllocatedInstance<PyRecordSetWriter>)},
+ {} /* Sentinel */
+};
+
+static PyType_Spec PyRecordSetWriterTypeSpec{
+ .name = "minifi_native.RecordSetWriter",
+ .basicsize = sizeof(PyRecordSetWriter),
+ .itemsize = 0,
+ .flags = Py_TPFLAGS_DEFAULT,
+ .slots = PyRecordSetWriterTypeSpecSlots
+};
+
+int PyRecordSetWriter::init(PyRecordSetWriter* self, PyObject* args, PyObject*) {
+ gsl_Expects(self && args);
+ PyObject* weak_ptr_capsule = nullptr;
+ if (!PyArg_ParseTuple(args, "O", &weak_ptr_capsule)) {
+ return -1;
+ }
+
+ auto record_set_writer = PyCapsule_GetPointer(weak_ptr_capsule, HeldTypeName);
+ if (!record_set_writer)
+ return -1;
+ self->record_set_writer_ = *static_cast<HeldType*>(record_set_writer);
+ return 0;
+}
+
+PyObject* PyRecordSetWriter::write(PyRecordSetWriter* self, PyObject* args) {
+ gsl_Expects(self && args);
+ auto record_set_writer = self->record_set_writer_.lock();
+ if (!record_set_writer) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading record set writer outside 'on_trigger'");
+ return nullptr;
+ }
+
+ PyObject* py_recordset = nullptr;
+ PyObject* script_flow_file = nullptr;
+ PyObject* py_session = nullptr;
+ if (!PyArg_ParseTuple(args, "OO!O!", &py_recordset, PyScriptFlowFile::typeObject(), &script_flow_file, PyProcessSessionObject::typeObject(), &py_session)) {
+ return nullptr;
+ }
+
+ if (!py_recordset) {
+ PyErr_SetString(PyExc_AttributeError, "Recordset is invalid!");
+ return nullptr;
+ }
+
+ const auto flow_file = reinterpret_cast<PyScriptFlowFile*>(script_flow_file)->script_flow_file_.lock();
+ if (!flow_file) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading FlowFile outside 'on_trigger'");
+ return nullptr;
+ }
+
+ const auto process_session = reinterpret_cast<PyProcessSessionObject*>(py_session)->process_session_.lock();
+ if (!process_session) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading ProcessSession outside 'on_trigger'");
+ return nullptr;
+ }
+
+ auto record_list = BorrowedList(py_recordset);
+ std::vector<core::Record> record_set;
+ for (size_t i = 0; i < record_list.length(); ++i) {
+ auto record_json_str = BorrowedStr(record_list[i].get()).toUtf8String();
+ rapidjson::Document document;
+ document.Parse<0>(record_json_str.c_str());
+ record_set.push_back(core::Record::fromJson(document));
+ }
+
+ record_set_writer->write(record_set, flow_file, process_session->getSession());
+ Py_RETURN_NONE;
+}
+
+PyTypeObject* PyRecordSetWriter::typeObject() {
+ static OwnedObject PyRecordSetWriterType{PyType_FromSpec(&PyRecordSetWriterTypeSpec)};
+ return reinterpret_cast<PyTypeObject*>(PyRecordSetWriterType.get());
+}
+
+} // namespace org::apache::nifi::minifi::extensions::python
+} // extern "C"
diff --git a/extensions/python/types/PyProcessContext.h b/extensions/python/types/PyRecordSetWriter.h
similarity index 55%
copy from extensions/python/types/PyProcessContext.h
copy to extensions/python/types/PyRecordSetWriter.h
index 8e3f515b4..661173874 100644
--- a/extensions/python/types/PyProcessContext.h
+++ b/extensions/python/types/PyRecordSetWriter.h
@@ -18,34 +18,28 @@
#include <memory>
-#include "core/ProcessContext.h"
-#include "PySSLContextService.h"
+#include "controllers/RecordSetWriter.h"
#include "../PythonBindings.h"
namespace org::apache::nifi::minifi::extensions::python {
-struct PyProcessContext {
- PyProcessContext() {}
- using HeldType = core::ProcessContext*;
- static constexpr const char* HeldTypeName = "PyProcessContext::HeldType";
+struct PyRecordSetWriter {
+ PyRecordSetWriter() {}
+ using HeldType = std::weak_ptr<core::RecordSetWriter>;
+ static constexpr const char* HeldTypeName = "PyRecordSetWriter::HeldType";
PyObject_HEAD
- HeldType process_context_;
+ HeldType record_set_writer_;
- static int init(PyProcessContext* self, PyObject* args, PyObject* kwds);
+ static int init(PyRecordSetWriter* self, PyObject* args, PyObject* kwds);
- static PyObject* getProperty(PyProcessContext* self, PyObject* args);
- static PyObject* getStateManager(PyProcessContext* self, PyObject* args);
- static PyObject* getControllerService(PyProcessContext* self, PyObject* args);
- static PyObject* getName(PyProcessContext* self, PyObject* args);
- static PyObject* getProperties(PyProcessContext* self, PyObject* args);
- static PyObject* yieldResources(PyProcessContext* self, PyObject* args);
+ static PyObject* write(PyRecordSetWriter* self, PyObject* args);
static PyTypeObject* typeObject();
};
namespace object {
template<>
-struct Converter<PyProcessContext::HeldType> : public HolderTypeConverter<PyProcessContext> {};
+struct Converter<PyRecordSetWriter::HeldType> : public HolderTypeConverter<PyRecordSetWriter> {};
} // namespace object
} // namespace org::apache::nifi::minifi::extensions::python
diff --git a/extensions/python/types/Types.h b/extensions/python/types/Types.h
index 0c14039f2..54769b44a 100644
--- a/extensions/python/types/Types.h
+++ b/extensions/python/types/Types.h
@@ -243,6 +243,15 @@ class List : public ReferenceHolder<reference_type> {
: ReferenceHolder<reference_type>(object) {
}
+ static OwnedList create() requires(reference_type == ReferenceType::OWNED) {
+ return OwnedList(PyList_New(0));
+ }
+
+ template<object::convertible T>
+ void append(T value) {
+ PyList_Append(this->ref_.get(), object::from(std::move(value)).releaseReference());
+ }
+
size_t length() {
return PyList_Size(this->ref_.get());
}
diff --git a/extensions/standard-processors/tests/unit/RecordSetTests.cpp b/extensions/standard-processors/tests/unit/RecordSetTests.cpp
new file mode 100644
index 000000000..e33822b42
--- /dev/null
+++ b/extensions/standard-processors/tests/unit/RecordSetTests.cpp
@@ -0,0 +1,177 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.c
+ */
+
+#include <numbers>
+#include <variant>
+#include <catch2/generators/catch_generators.hpp>
+
+#include "unit/Catch.h"
+#include "unit/TestBase.h"
+#include "unit/TestRecord.h"
+#include "utils/TimeUtil.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::standard::test {
+
+TEST_CASE("Test JSON serialization of a RecordField") {
+ {
+ minifi::core::RecordField field{1};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetUint64() == 1);
+ }
+ {
+ minifi::core::RecordField field{-1};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetInt64() == -1);
+ }
+ {
+ minifi::core::RecordField field{false};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetBool() == false);
+ }
+ {
+ minifi::core::RecordField field{'a'};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetString() == std::string("a"));
+ }
+ {
+ minifi::core::RecordField field{1.2};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetDouble() == 1.2);
+ }
+ {
+ minifi::core::RecordField field{std::string("hello")};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetString() == std::string("hello"));
+ }
+ {
+ std::chrono::time_point test_time = utils::timeutils::parseDateTimeStr("2021-09-01T12:34:56Z").value();
+ minifi::core::RecordField field{test_time};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value.GetString() == std::string("2021-09-01T12:34:56Z"));
+ }
+ {
+ minifi::core::RecordObject obj;
+ obj.emplace("key1", std::make_unique<minifi::core::RecordField>(1));
+ obj.emplace("key2", std::make_unique<minifi::core::RecordField>(std::string("hello")));
+ minifi::core::RecordField field{std::move(obj)};
+ rapidjson::Document doc;
+ auto value = field.toJson(doc.GetAllocator());
+ CHECK(value["key1"].GetUint64() == 1);
+ CHECK(value["key2"].GetString() == std::string("hello"));
+ }
+
+ {
+ minifi::core::RecordField field1{-1};
+ minifi::core::RecordField field2{true};
+ std::vector<minifi::core::RecordField> arr;
+ arr.push_back(std::move(field1));
+ arr.push_back(std::move(field2));
+ minifi::core::RecordField array_field{std::move(arr)};
+ rapidjson::Document doc;
+ auto value = array_field.toJson(doc.GetAllocator());
+ CHECK(value.GetArray()[0].GetInt64() == -1);
+ CHECK(value.GetArray()[1].GetBool() == true);
+ }
+}
+
+TEST_CASE("Test JSON serialization of a Record") {
+ minifi::core::Record record;
+ record.emplace("key1", minifi::core::RecordField{1});
+ record.emplace("key2", minifi::core::RecordField{std::string("hello")});
+ record.emplace("key3", minifi::core::RecordField{true});
+ record.emplace("key4", minifi::core::RecordField{1.2});
+ std::chrono::time_point test_time = utils::timeutils::parseDateTimeStr("2021-09-01T12:34:56Z").value();
+ record.emplace("key5", minifi::core::RecordField{test_time});
+
+ minifi::core::RecordField field1{-1};
+ minifi::core::RecordField field2{true};
+ std::vector<minifi::core::RecordField> arr;
+ arr.push_back(std::move(field1));
+ arr.push_back(std::move(field2));
+ record.emplace("key6", minifi::core::RecordField{std::move(arr)});
+
+ minifi::core::RecordObject subobj;
+ subobj.emplace("subkey3", std::make_unique<minifi::core::RecordField>(1));
+ subobj.emplace("subkey4", std::make_unique<minifi::core::RecordField>(std::string("subhello")));
+ minifi::core::RecordObject obj;
+ obj.emplace("subkey1", std::make_unique<minifi::core::RecordField>(-2));
+ obj.emplace("subkey2", std::make_unique<minifi::core::RecordField>(std::move(subobj)));
+ record.emplace("key7", minifi::core::RecordField{std::move(obj)});
+
+ rapidjson::Document doc = record.toJson();
+ CHECK(doc["key1"].GetUint64() == 1);
+ CHECK(doc["key2"].GetString() == std::string("hello"));
+ CHECK(doc["key3"].GetBool() == true);
+ CHECK(doc["key4"].GetDouble() == 1.2);
+ CHECK(doc["key5"].GetString() == std::string("2021-09-01T12:34:56Z"));
+ CHECK(doc["key6"].GetArray()[0] == -1);
+ CHECK(doc["key6"].GetArray()[1] == true);
+ CHECK(doc["key7"]["subkey1"].GetInt64() == -2);
+ CHECK(doc["key7"]["subkey2"]["subkey3"].GetUint64() == 1);
+ CHECK(doc["key7"]["subkey2"]["subkey4"].GetString() == std::string("subhello"));
+}
+
+TEST_CASE("Test Record deserialization from JSON") {
+ std::string json_str = R"(
+{
+ "number": 1,
+ "string": "hello",
+ "bool": false,
+ "double": 1.2,
+ "array": [1.1, false],
+ "time_point": "2021-09-01T12:34:56Z",
+ "obj": {
+ "number2": 2,
+ "message": "mymessage"
+ }
+}
+ )";
+ rapidjson::Document doc;
+ doc.Parse<0>(json_str);
+ auto record = minifi::core::Record::fromJson(doc);
+ CHECK(record.at("number") == minifi::core::RecordField{1});
+ CHECK(record.at("string") == minifi::core::RecordField{std::string("hello")});
+ CHECK(record.at("bool") == minifi::core::RecordField{false});
+ CHECK(record.at("double") == minifi::core::RecordField{1.2});
+ std::chrono::time_point test_time = utils::timeutils::parseDateTimeStr("2021-09-01T12:34:56Z").value();
+ CHECK(record.at("time_point") == minifi::core::RecordField{test_time});
+
+ minifi::core::RecordObject subobj;
+ subobj.emplace("number2", std::make_unique<minifi::core::RecordField>(2));
+ subobj.emplace("message", std::make_unique<minifi::core::RecordField>(std::string("mymessage")));
+ minifi::core::RecordField obj_field{std::move(subobj)};
+ CHECK(record.at("obj") == obj_field);
+
+ minifi::core::RecordField field1{1.1};
+ minifi::core::RecordField field2{false};
+ std::vector<minifi::core::RecordField> arr;
+ arr.push_back(std::move(field1));
+ arr.push_back(std::move(field2));
+ minifi::core::RecordField array_field{std::move(arr)};
+ CHECK(record.at("array") == array_field);
+}
+
+} // namespace org::apache::nifi::minifi::standard::test
diff --git a/libminifi/include/core/Record.h b/libminifi/include/core/Record.h
index 7e2a5a89d..684da33ac 100644
--- a/libminifi/include/core/Record.h
+++ b/libminifi/include/core/Record.h
@@ -23,8 +23,9 @@
#include "RecordField.h"
-namespace org::apache::nifi::minifi::core {
+#undef GetObject // windows.h #defines GetObject = GetObjectA or GetObjectW, which conflicts with rapidjson
+namespace org::apache::nifi::minifi::core {
class Record final {
public:
@@ -55,6 +56,25 @@ class Record final {
bool operator==(const Record& rhs) const = default;
+ rapidjson::Document toJson() const {
+ rapidjson::Document doc;
+ auto& alloc = doc.GetAllocator();
+ rapidjson::Value obj(rapidjson::kObjectType);
+ for (const auto& [key, field] : fields_) {
+ obj.AddMember(rapidjson::Value(key.c_str(), alloc), field.toJson(alloc), alloc);
+ }
+ doc.Swap(obj);
+ return doc;
+ }
+
+ static Record fromJson(const rapidjson::Document& document) {
+ Record record;
+ for (const auto& member : document.GetObject()) {
+ record.emplace(member.name.GetString(), RecordField::fromJson(member.value));
+ }
+ return record;
+ }
+
private:
std::unordered_map<std::string, RecordField> fields_;
};
diff --git a/libminifi/include/core/RecordField.h b/libminifi/include/core/RecordField.h
index ae89e67c2..ca6bf2e71 100644
--- a/libminifi/include/core/RecordField.h
+++ b/libminifi/include/core/RecordField.h
@@ -24,6 +24,7 @@
#include <variant>
#include <chrono>
#include <concepts>
+#include "rapidjson/document.h"
namespace org::apache::nifi::minifi::core {
@@ -78,6 +79,8 @@ struct RecordField {
~RecordField() = default;
+ rapidjson::Value toJson(rapidjson::Document::AllocatorType& alloc) const;
+ static RecordField fromJson(const rapidjson::Value& value);
bool operator==(const RecordField& rhs) const = default;
diff --git a/libminifi/src/core/RecordField.cpp b/libminifi/src/core/RecordField.cpp
new file mode 100644
index 000000000..32042618d
--- /dev/null
+++ b/libminifi/src/core/RecordField.cpp
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "core/RecordField.h"
+
+#include "utils/GeneralUtils.h"
+#include "utils/TimeUtil.h"
+
+namespace org::apache::nifi::minifi::core {
+
+rapidjson::Value RecordField::toJson(rapidjson::Document::AllocatorType& allocator) const {
+ rapidjson::Value value;
+ std::visit(utils::overloaded {
+ [&value, &allocator](const std::string& str) {
+ value.SetString(str.c_str(), allocator);
+ },
+ [&value](int64_t i64) {
+ value.SetInt64(i64);
+ },
+ [&value](uint64_t u64) {
+ value.SetUint64(u64);
+ },
+ [&value](double d) {
+ value.SetDouble(d);
+ },
+ [&value](bool b) {
+ value.SetBool(b);
+ },
+ [&value, &allocator](const std::chrono::system_clock::time_point& time_point) {
+ value.SetString(utils::timeutils::getDateTimeStr(std::chrono::time_point_cast<std::chrono::seconds>(time_point)).c_str(), allocator);
+ },
+ [&value, &allocator](const RecordArray& arr) {
+ value.SetArray();
+ for (const auto& elem : arr) {
+ rapidjson::Value elem_value = elem.toJson(allocator);
+ value.PushBack(elem_value, allocator);
+ }
+ },
+ [&value, &allocator](const RecordObject& obj) {
+ value.SetObject();
+ for (const auto& [key, boxed_field] : obj) {
+ rapidjson::Value keyValue;
+ keyValue.SetString(key.c_str(), allocator);
+
+ rapidjson::Value fieldValue = boxed_field.field->toJson(allocator);
+ value.AddMember(keyValue, fieldValue, allocator);
+ }
+ }
+ }, value_);
+
+ return value;
+}
+
+RecordField RecordField::fromJson(const rapidjson::Value& value) {
+ if (value.IsString()) {
+ std::string str_value = value.GetString();
+ if (auto test_time = utils::timeutils::parseDateTimeStr(str_value)) {
+ return RecordField{std::chrono::time_point{*test_time}};
+ }
+ return RecordField{str_value};
+ } else if (value.IsInt64()) {
+ return RecordField{value.GetInt64()};
+ } else if (value.IsUint64()) {
+ return RecordField{value.GetUint64()};
+ } else if (value.IsDouble()) {
+ return RecordField{value.GetDouble()};
+ } else if (value.IsBool()) {
+ return RecordField{value.GetBool()};
+ } else if (value.IsArray()) {
+ RecordArray arr;
+ for (const auto& elem : value.GetArray()) {
+ arr.push_back(RecordField::fromJson(elem));
+ }
+ return RecordField{std::move(arr)};
+ } else if (value.IsObject()) {
+ RecordObject obj;
+ for (const auto& member : value.GetObject()) {
+ obj.emplace(member.name.GetString(), BoxedRecordField{std::make_unique<RecordField>(RecordField::fromJson(member.value))});
+ }
+ return RecordField{std::move(obj)};
+ } else {
+ throw std::runtime_error("Invalid JSON value type");
+ }
+}
+
+} // namespace org::apache::nifi::minifi::core