# Product Flow The product flow facilitates the transfer of product information like attributes, name, sku from ERP systems to Omnitron. Flow will create new attribute values, update existing products or create new ones for different languages. The flow can operate in two ways: as inbound or outbound. Inbound product flow means that the flow triggers itself at specific intervals set by the user,such as once every minute, every 1 hour, every day at midnight, every 5th day of a month etc. to read data from the ERP system and write it to Omnitron, ensuring Omnitron has the latest product data. Outbound product flow, on the other hand, involves data being pushed directly to the flow from an external source and then written to Omnitron. This dual capability ensures seamless communication and accurate product management across the entire system. ### Product Flow Types​ * **Inbound Flow:** Indicates that the flow will read data from the ERP system. A cron schedule is set up to ensure it runs, for example, every 5 minutes. By default entegrator will use a standard REST API structure to read data from ERP systems. Without any customization the request will be made as follows: | Property | Value | | --------------------------- | ------------------------------------------------------------------------------------------------- | | URL Structure | \/?modified\_date\_\_gt=modified\_date\_\_gt=2025-12-21T12:00:30.000000\&language=tr | | HTTP Method | GET | | Content-Type (HTTP Header) | application/json | | Authorization (HTTP Header) | \ | * **Outbound Flow:** Signifies that product data will be posted to the flow from the ERP system, skipping the step of reading data from the ERP system. In outbound flows, triggers will be configured to send a POST request to the URL specified in the "Outbound Request URL" setting under the Configuration card. By default the content of the POST request will should include JSON data containing details such as SKU, language, attributes, etc. for each product. Sample data structure for both inbound and outbound flows: ```json [ { "name": "LS VELOUR BOMBER", "base_code": "T41006409X", "sku": "5643981820", "is_active": false, "language": "en-en", "attributes": { "ColorId": "Y0", "ColorDesc": "BLACK", "SizeId": "14", "SizeDesc": "14" } }, { "name": "LS VELOUR BOMBER", "base_code": "T41006409X", "sku": "5643981821", "is_active": false, "language": "en-en", "attributes": { "ColorId": "Y1", "ColorDesc": "WHITE", "SizeId": "14", "SizeDesc": "14" } } ] ``` | parameter | data type | required field | example | description | additional notes | | ---------- | --------- | -------------- | ------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | | name | string | yes | LS VELOUR BOMBER | Product name | | | base\_code | string | yes | T41006409X | Model Number | Variants are based on the data of this field | | sku | string | yes | 5643981820 | Unique representation of the product | | | is\_active | boolean | yes | true | If product is active | The parameter that affects the visibility of the product in channels. | | attributes | object | yes |

{
"ColorId": "Y0",
"ColorDesc": "BLACK",
"SizeId": "14",
"SizeDesc": "14"
}

| Product properties that vary from company to company are listed under this field. There may also be additional information required to be included in the product. Key and value should be set. From the Omnitron panel you can also add feature information about the product. Product description, color, size etc. Features can be shown as examples. | All specific product specifications defined in the company can be submitted. | | language | string | yes | en-en | Language | | ### Flow Steps​ * **Login Step:** Handles logging into Omnitron and (if configured) the ERP system. If an error occurs during this step, the details are logged. * **Read Data From ERP Step:** For inbound flows, product queries are made from the ERP system. Outbound flows proceed directly to the next step. * **Script Step (Optional):** Transformation operations on the data are performed using Python, if required. * **Mapping Step (Optional):** Transformation operations on the data are performed using the Jolt Transform library. Further details can be found at the [Jolt Transform](https://jolt-demo.appspot.com/#inception) website. * **Write Data to Omnitron Step:** The incoming data is divided into rows and written to Omnitron. Logs are created for each row. In case of an error, details of the error are logged (e.g., the relevant product attribute is not correct). Successful operations are also logged. ### Detailed Flow Designer Settings​ #### Configuration Card​
* **Flow Trigger Parameters:** Configures the key values to be used for single and date-based queries from ERP. `sku` and `modified_date__gt` are created by default and are required values. These settings are used in the trigger page to use specific parameters while creating the trigger url. `modified_date__gt` value is also used while creating automatic inbound request urls and `sku` value is used on “Fetch Missing” tasks for querying missing SKU's on ERP. ***Adding a new Query Parameter:***
***Using newly added parameters:***
***Effects of the custom parameters:*** * Using with SKU: * [*http://example.com/api/product?language=TR\&sku=examplesku0\&custom\_parameter=5*](http://example.com/api/product?language=TR\&sku=examplesku0\&custom_parameter=5) * Using with Modified Date: * [*http://example.com/api/product?language=TR\&modified\_date\_\_gt=2024-10-01T00:00:00.000Z\&custom\_parameter=5*](http://example.com/api/product?language=TR\&modified_date__gt=2024-10-01T00:00:00.000Z\&custom_parameter=5) * **Omnitron API Settings** * **Will Pre-Products be used?:** This should be set to **true** if the target Omnitron system is using pre-products. * **Domain URL:** The domain URL for Omnitron. Example: . * **Timezone Settings:** Timezone used for date-based queries from ERP, this value is used with inbound task requests to modify `modified_date__gt` value for a specific timezone. The date time format is following: `yyyy-MM-dd'T'HH:mm:ss.SSSSSS`, ex: `2024-12-31T13:30:59.000001`
* **Language:** Languages that will be queried from ERP, separated by a comma. When a value like "TR,EN" is used with an Inbound flow each time the flow runs 2 executions will start one for each language and it will create a query using language key, in this example it would create 2 executions. #### Read Data From ERP Card​ * **Add Extra Headers as Dict Format:** During the product query from ERP, additional headers to be sent.
* **Extra Params:** Additional parameters used when ERP request is made, it must be a valid dict if a GET request is made which will be used to send query parameters in the url or it can be used on a POST request in which the extra params value will be used to be sent in request body. * **Language Key:** Key will be used while sending language values set in the Configuration step in the request. * **Product API URL:** URL for reading data from the ERP system.
**Dynamic URL Usage:​** Dynamic URLs are used when the url structure, path changes based on different conditions or if the URL doesn’t fit a standard REST API structure. In the example below, the URL changes if the request is going to be a “date” based query or “sku” based query and at the same time the API requires the ERP token to be in the URL itself. Using [Nifi Expression](https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html) language it is possible to change the request URL in the run time based on conditions and expressions written in the URL field. ``` ${http.query.string:isEmpty():not(): ifElse( ${http.query.param.sku:isEmpty():not() :ifElse( ${literal("http://127.0.0.1:12345/(S({TOKEN}))/service/run?{'name': 'Akinon', 'Parameters': [{'Name': 'Barcodes', 'Value': '{SKU}'}]}") :replace("{SKU}", ${http.query.param.sku}) :replace("{TOKEN}",${token.erp}) }, ${literal("http://127.0.0.1:12345/(S({TOKEN}))/service/run?{'name': 'Akinon', 'Parameters': [{'Name': 'Date', 'Value': '{MODIFIED_DATE}'}]}") :replace("{MODIFIED_DATE}", ${http.query.param.modified_date__gt:replaceAll("(\.\d*)?Z?$","")}) :replace("{TOKEN}",${token.erp}) } )}, ${literal("http://127.0.0.1:12345/(S({TOKEN}))/service/run?{'name': 'Akinon', 'Parameters': [{'Name': 'Date', 'Value': '{MODIFIED_DATE}'}]}") :replace("{MODIFIED_DATE}", ${last.request:replaceAll("(\.\d*)?Z?$","")}) :replace("{TOKEN}",${token.erp}) } ) } ``` **HTTP Method:** `GET` or `POST` used for ERP requests. **ERP Response Data Type:** * If the ERP returns an XML response instead of JSON, this setting can be used to parse and transform ERP response to a JSON to use in further steps. **Pagination:** Pagination in Integrator works by creating a new execution for each page, using the same initial parameters and continuing with the configured pagination method. The process continues until the response no longer contains valid data — either the response body is empty, or the status code is not in the 2xx range. This approach causes Integrator to make one extra request per task to ensure all possible data has been retrieved. A new task will not begin until the currently running task has fetched all available pages. The "last request date" parameter — typically used as modified\_date\_\_gt — is only updated after Integrator has successfully fetched all pages and the final response returns empty data. At that point, a new scheduled task can start using an updated "last request date" value, which corresponds to the timestamp just before the first request of the previous successful task. {% hint style="info" %} "Status codes to be assumed as empty page for pagination" setting can be configured to make Integrator consider a certain status code as "Empty" response resulting in an overall successful execution. This is useful if API endpoint used responds with a non 2xx http status code when a empty page is queried. {% endhint %} * Offset pagination is a method where a fixed number of items are retrieved from a data source in each request, starting from a specified offset. This offset indicates the position from which to begin fetching data. This pagination type is useful when pages are constructed by skipping an amount of data, for example; * First page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&limit=50&skip=0` * Second page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&limit=50&skip=50` * Third page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&limit=50&skip=100`
* Seek pagination is a method where the app fetches pages sequentially by incrementing a numeric page parameter with each request. Each page contains a fixed number of items. This is often simpler than offset pagination, as it directly references a page number rather than calculating offsets. This pagination type is useful when the API expects explicit page numbers, for example: * First page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&page=1&limit=50` * Second page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&page=2&limit=50` * Third page: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&page=3&limit=50`
* Next field pagination is a method where the API response itself includes a URL or token pointing to the next page of data. The app follows this URL to retrieve the next set of results, continuing until there are no more pages. This is useful when the server controls how to continue pagination, for example: ```json { "results": [ /* items */ ], "next": "/api/data/?cursor=abc123" } ``` The app will automatically use the next field to make the subsequent request: * First request: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000` * Follow-up request: `/api/data/?modified_date__gt=2025-07-22T10:30:27.000000&cursor=abc123` * And so on, until next is null or not present.
#### Script Card (Optional)​ This is a base script that can be updated to enable script functionality within the process step. Scripts can read incoming data from the `inputStream`, where the variable `input_text` contains a JSON string. This string should be parsed using Python's `json` library. Any outgoing data is written using the `outputStream.write()` method after converting the relevant Python object back into a JSON string. Additionally, the script allows for the use of attributes, which provide supplementary information tied to the current execution process. These attributes can be freely accessed or modified throughout the script. For example, the `get_attribute()` function is used to read attribute values, while the `session.putAttribute()` method is used to write new string attribute values. Each attribute consists of a key-value pair, where the `key` uniquely identifies the attribute, and the `value` can be referenced in subsequent steps. Attributes with keys starting with the prefix `log.`will be automatically logged at the end of the execution if a log file is generated, ensuring that important information is captured and available for later review.
**Example Script:** ``` import json import traceback from java.nio.charset import StandardCharsets from org.apache.commons.io import IOUtils from org.apache.nifi.processor.io import StreamCallback def get_attribute(flow_file, attr_name): all_var = flow_file.getAttribute("allVar") if all_var: all_attributes = json.loads(all_var) return all_attributes.get(attr_name, None) return flow_file.getAttribute(attr_name) class TransformCallback(StreamCallback): def __init__(self, flowFile): self.flowFile = flowFile self.omnitronToken = get_attribute(flowFile, "token.omnitron") self.erpToken = get_attribute(flowFile, "token.erp") def process(self, inputStream, outputStream): input_text = IOUtils.toString(inputStream, StandardCharsets.UTF_8) input_obj = json.loads(input_text) # Transform content output_obj = self.transform(input_obj) # Write output content outputStream.write(bytearray(json.dumps(output_obj, indent=4).encode('utf-8'))) def transform(self, output_obj): # Transform content return output_obj flowFile = session.get() if flowFile != None: try: flowFile = session.write(flowFile, TransformCallback(flowFile)) # Finish by transferring the FlowFile to an output relationship session.transfer(flowFile, REL_SUCCESS) except: var = traceback.format_exc() session.putAttribute(flowFile, 'log.script.error', str(var)) session.transfer(flowFile, REL_FAILURE) session.commit() ``` **Script Testing:** The ERP response is placed in the input field, and the script's result along with the written attributes can be viewed in the result field. When the test script is run, attributes like `log.script.error` will appear under the "attributes" section in the result panel and should be referenced if any exceptions occur during execution. The script's output content will be found under the `result_data` value. The **+ Add Attributes** button can be used to simulate incoming, readable, attributes for the script. This feature is useful when an attribute value is read and used in the script. Note that attributes added through the button are only for testing purposes and will not be saved in the flow information—they will be removed when the page is refreshed. #### Mapping Card (Optional)​
Details can be found at the [Jolt Transform](https://jolt-demo.appspot.com/#inception) website. Since this is the final step where the input data can be transformed, the flow requires the following output to be generated for correct functionality. In product flows, additional Jolt specifications are applied by the Integrator during execution, after the mapping specification added by the user. The expected output value includes both the mapping specification provided by the user and the one applied by the Integrator. The output of the mapping step must include the "products" value as an array. Each product in this array will be sent separately to Omnitron for either updating or creating the product. Additionally, the "attributes\_kwargs" is required to create dropdown values in Omnitron. The "dropdown\_info" in the mapping defines the dropdown attributes for the product, with the "key" representing the name of the attribute itself. The "label" and "value" should correspond to key names in the "pre\_attributes" dict. In the example below, "attribute\_value\_1\_value" and "attribute\_value\_1\_label" are mapped to `pre_attributes.color_value` and `pre_attributes.color_label`. In the `dropdown_info`, the "erp\_color" dropdown attribute is defined with its "value" set to "color\_value" and its "label" set to "color\_label". When the extra specifications added by the integrator system are applied, the label and value are looked up in "pre\_attributes" and used to create the "attributes\_kwargs". {% hint style="info" %} The *pre\_attributes* in this context is not related to Omnitron's own "pre\_attributes"; it is simply named the same. {% endhint %} **Example Input:** ``` [ { "is_active": true, "base_code": "123", "language": "tr-tr", "name": "TSHIRT", "attributes": { "size": "XL", "attribute_value_1_value": "1", "attribute_value_1_label": "MAVI" }, "sku": "1" } ] ``` **Example Mapping:** ``` [ { "operation": "shift", "spec": { "*": { "extra_attributes": "[&1].extra_attributes", "product_type": "[&1].product_type", "sku": "[&1].sku", "base_code": "[&1].base_code", "name": "[&1].name", "is_active": "[&1].is_active", "language": "[&1].language", "@(0,attributes)": "[&1].pre_attributes", "attributes": { "size": "[&2].attributes.erp_size", "attribute_value_1_label": "[&2].pre_attributes.color_label", "attribute_value_1_value": "[&2].pre_attributes.color_value" } } } }, { "operation": "default", "spec": { "*": { "dropdown_info": { "erp_color": { "key": "erp_color", "value": "color_value", "label": "color_label" } } } } } ] ``` **Extra specification added by Integrator:** The following JOLT specification is added by Integrator regardless of the user’s own mapping specifications. The user can not change or view the following specification, this is included in the documentation for further references only. ``` [ ${mapping code added by user}, { "operation": "default", "spec": { "*": { "extra_attributes": {}, "product_type": "${omnitron.has_pre_product:equals('Yes'):ifElse('-1','0')}" } } }, { "operation": "shift", "spec": { "*": { "@": "products.[&1]", "dropdown_info": { "*": { "@(key)": { "*": { "$": "attributes_kwargs.[&5].[#4].key", "@(4,language)": "attributes_kwargs.[&5].[#4].language", "@(2,value)": { "*": { "@(6,pre_attributes.&)": "products.[&7].attributes.&2" } } } }, "@(value)": { "*": { "@(4,pre_attributes.&)": "attributes_kwargs.[&5].[#4].value" } }, "@(label)": { "*": { "@(4,pre_attributes.&)": "attributes_kwargs.[&5].[#4].label" } } } } } } }, { "operation": "remove", "spec": { "products": { "*": { "dropdown_info": "" } } } }, { "operation": "shift", "spec": { "products": "products", "@(0,attributes_kwargs)": { "*": { "*": "attributes_kwargs[]" } } } }, { "operation": "modify-default-beta", "spec": { "attributes_kwargs": { "*": { "a": "=concat(@(1,label),'',@(1,value),'',@(1,key),'',@(1,language))" } } } }, { "operation": "modify-overwrite-beta", "spec": { "products": { "*": { "attributes": { "*": "=toString" } } }, "attributes_kwargs": { "*": { "value": "=toString", "label": "=toString" } } } }, { "operation": "shift", "spec": { "products": "products", "attributes_kwargs": { "*": { "a": { "@(1)": "attributes_kwargs.@(2,a)" } } } } }, { "operation": "cardinality", "spec": { "attributes_kwargs": { "*": { "@": "ONE" } }, "products": { "*": { "attributes": { "*": { "@": "ONE" } } } } } }, { "operation": "shift", "spec": { "products": "products", "attributes_kwargs": { "*": { "@": "attributes_kwargs[]" } } } }, { "operation": "remove", "spec": { "products": { "*": { "pre_attributes": "" } }, "attributes_kwargs": { "*": { "a": "" } } } } ] ``` **Expected Output:** This output is the result of both the user's own mapping specifications and the one applied by the Integrator. ``` { "products": [ { "sku": "1", "base_code": "123", "name": "TSHIRT", "is_active": true, "language": "tr-tr", "attributes": { "erp_size": "XL", "erp_color": "1" }, "extra_attributes": {}, "product_type": "0" } ], "attributes_kwargs": [ { "key": "erp_color", "language": "tr-tr", "value": "1", "label": "MAVI" } ] } ``` **Mapping Testing:** The response from the script or the response of the ERP request if the script step is not used, is placed in the input field and the result of the mapping is viewed in the result field.​ #### `POST` or `PATCH` Write Data to Omnitron **Exclude Fields:**
Used for removing product information from PATCH data on products that will be updated. JOLT “remove” operation is used for this field. The above example makes sure fields like, name, erp\_season attribute etc. are **not updated** for products that already exist in Omnitron. That way if any of the fields are manually updated or changed in Omnitron the update coming from ERP will not overwrite the fields. Integrator will use different API endpoints, * If “Pending Products” are available in the target omnitron and **"Will Pre-Products be used?"** setting in the flow configuration is set to “true” following requests will be made: * `POST` **Path:** `/api/v1/erp_products/` or `/api/v1/erp_pre_products/` (If the product does not already exist on Omnitron.) * `PATCH` **Path:** `/api/v1/erp_products//` or `/api/v1/erp_products//` (If the product already exists on Omnitron.) * If **"Will Pre-Products be used?"** setting in the flow configuration is set to “false” following requests will be made: * `POST` **Path:** `/api/v1/products/` (If the product does not already exist on Omnitron.) * `PATCH` **Path:** `/api/v1/products//` (If the product already exists on Omnitron.) **Example POST request from Integrator to Omnitron:** ``` { "sku": "1", "base_code": "123", "name": "TSHIRT", "is_active": true, "language": "tr-tr", "attributes": { "erp_size": "XL", "erp_color": "1" }, "extra_attributes": {}, "product_type": "0" } ``` ### Outbound Flow​ In outbound flows, triggers will be configured to send a POST request to the URL specified in the "Outbound Request URL" setting under the Configuration card. The content of the POST request will include JSON data containing details such as SKU, language, attributes, etc. for each product.
**Example Payload:** ```json [ { "is_active": true, "base_code": "123", "language": "tr-tr", "name": "TSHIRT", "attributes": { "size": "XL", "attribute_value_1_value": "1", "attribute_value_1_label": "MAVI" }, "sku": "1" }, { "is_active": true, "base_code": "123", "language": "tr-tr", "name": "TSHIRT", "attributes": { "size": "XL", "attribute_value_1_value": "2", "attribute_value_1_label": "KIRMIZI" }, "sku": "2" }, { "is_active": true, "base_code": "123", "language": "tr-tr", "name": "TSHIRT", "attributes": { "size": "XL", "attribute_value_1_value": "3", "attribute_value_1_label": "TURUNCU" }, "sku": "3" } ] ``` **Example response (HTTP 200):** ```json { "execution_id": "4054eae0-6165-43d3-9305-cf33b185086b" } ``` {% hint style="info" %} Before sending data to this outbound flow, you must log in with an Integrator user and obtain an authorization token. The login method is documented in the [Login API documentation](https://apidocs.akinon.com/integrator/authentication), and the token must be included in the `Authorization` header as `Bearer `. {% endhint %} --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.akinon.com/tutorials/integrator/flows/product-flow.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.