Order Status Flow

The Order Status Flow facilitates the transfer of order information, such as invoice numbers, URLs, order statuses, and shipment details, from ERP systems to Omnitron. The flow updates existing orders with the latest information or creates new entries as required. The flow can operate in two ways: as inbound or outbound.

Inbound order status 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 update it in Omnitron, ensuring Omnitron reflects the latest order information. Outbound Order Status Flow, on the other hand, involves data being pushed directly to the flow from an external source and then updated in Omnitron.

This dual capability ensures seamless communication and accurate order management across the entire system.

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

• Outbound Flow: Signifies that order status data will be posted to the flow from the ERP system, skipping the step of reading data from the ERP system.

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, order queries are made to 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 website.

• Write Data to Omnitron Step: The incoming data is divided into rows, in this step it is possible write another Python scripts, where unlike the previous “Script Step” the incoming data will have only one order object used mostly to do individual transformations, requests per order object and then written to Omnitron. Logs are created for each row. In case of an error, details of the error are logged (e.g., the relevant order item status is not correct or can not be updated). 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. “number” and “modified_dategt” 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_dategt” value is also used while creating automatic inbound request urls.

Adding a new Query Parameter:

Using newly added parameters:

• Effects of the custom parameters:

  • Using with Order Number:

    • http://example.com/api/orders?number=exampleorder&custom_parameter=customvalue
  • Using with Modified Date:

    • http://example.com/api/orders?modified_date__gt=2024-10-01T00:29:59.000Z&custom_parameter=customvalue
• Omnitron API Settings

  • Domain URL: Omnitron's domain URL, Example: https://demo.omnitron.akinon.net/.
  • 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

    

Read Data From ERP Card​

• Add Extra Headers as Dict Format: During the order 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.

• Order Query/Status 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 “order number” based query and at the same time the API requires the ERP token to be in the URL itself. Using Nifi Expression 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': 'Order', 'Value': '{NUMBER}'}]}")
                    :replace("{NUMBER}", ${http.query.param.number})                             
                    :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:

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

• Seek pagination relies on a unique identifier or a specific value to retrieve subsequent sets of data. Instead of using offsets, it uses a marker or token indicating where the previous set of data ended.

  
• Next field pagination includes a "next" field or token in each server response that points to the next page of data.

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.

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 website. Because after this step the input data will be separated to different order rows, the flow expects the following output to be an array of order objects.

Example Input:

[
    {
        "orderitems": [
            {
                "product": "product",
                "status": 450,
                "e_archive_url": "https://example.com/invoice/order1",
                "invoice_date": "2024-10-01T12:30:59",
                "id": 1,
                "invoice_number": "invoicenumber1"
            }
        ],
        "order": "ordernumber1"
    },
    {
        "orderitems": [
            {
                "product": "product",
                "shipping_date": "2024-10-01T12:35:59",
                "status": 500,
                "shipping_company": "cargocompany",
                "e_archive_url": "https://example.com/invoice/order2",
                "invoice_date": "2024-10-01T12:30:59",
                "id": 2,
                "invoice_number": "invoicenumber2"
            }
        ],
        "order": "ordernumber2"
    },
    {
        "orderitems": [
            {
                "product": "product",
                "shipping_date": "2024-10-01T12:35:59",
                "shipping_company": "cargocompany",
                "status": 500,
                "e_archive_url": "https://example.com/invoice/order2",
                "invoice_date": "2024-10-01T12:30:59",
                "id": 3,
                "invoice_number": "invoicenumber3"
            },
            {
                "product": "product",
                "shipping_date": "2024-10-01T12:35:59",
                "shipping_company": "cargocompany",
                "status": 500,
                "e_archive_url": "https://example.com/invoice/order2",
                "invoice_date": "2024-10-01T12:30:59",
                "id": 4,
                "invoice_number": "invoicenumber3"
            }
        ],
        "order": "ordernumber3"
    }
]

Example Mapping:

[
  {
    "operation": "shift",
    "spec": {
    "*": {
        "orderitems": "[&1].orderitem_set",
        "order": "[&1].order"
    }
    }
  }
]

Expected Output:

This output is the result of both the user's mapping specifications.

[
  {
    "orderitem_set": [
    {
        "product": "product",
        "status": 450,
        "e_archive_url": "https://example.com/invoice/order1",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 1,
        "invoice_number": "invoicenumber1"
    }
    ],
    "order": "ordernumber1"
  },
  {
    "orderitem_set": [
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "status": 500,
        "shipping_company": "cargocompany",
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 2,
        "invoice_number": "invoicenumber2"
    }
    ],
    "order": "ordernumber2"
  },
  {
    "orderitem_set": [
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 3,
        "invoice_number": "invoicenumber3"
    },
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 4,
        "invoice_number": "invoicenumber3"
    }
    ],
    "order": "ordernumber3"
  }
]

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.

PATCH Write Data to Omnitron​

Before order items are updated using order status flow, another script step is available, this script is optional and runs with individual order objects for input and expects the final order object. For order status flow, order value(should be equal to omnitron order number used for logging), orderitem_set array and in each orderitem object in orderitem_set “id” values are required.

Path: /api/i1/order_items/bulk_status_update/

Example PATCH request from Integrator to Omnitron:

This body can have more and less values to update based on what Omnitron order items accept and what is really updated.

  {
    "orderitem_set": [
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 3,
        "invoice_number": "invoicenumber3"
    },
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 4,
        "invoice_number": "invoicenumber3"
    }
    ],
    "order": "ordernumber3"
  }

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 orders to be updated, order numbers, orderitems, status, etc. for each row.

Example Payload:

[
  {
    "orderitem_set": [
    {
        "product": "product",
        "status": 450,
        "e_archive_url": "https://example.com/invoice/order1",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 1,
        "invoice_number": "invoicenumber1"
    }
    ],
    "order": "ordernumber1"
  },
  {
    "orderitem_set": [
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "status": 500,
        "shipping_company": "cargocompany",
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 2,
        "invoice_number": "invoicenumber2"
    }
    ],
    "order": "ordernumber2"
  },
  {
    "orderitem_set": [
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 3,
        "invoice_number": "invoicenumber3"
    },
    {
        "product": "product",
        "shipping_date": "2024-10-01T12:35:59",
        "shipping_company": "cargocompany",
        "status": 500,
        "e_archive_url": "https://example.com/invoice/order2",
        "invoice_date": "2024-10-01T12:30:59",
        "id": 4,
        "invoice_number": "invoicenumber3"
    }
    ],
    "order": "ordernumber3"
  }
]

The authorization token will be obtained by logging in with the Integrator user and must be included in the request headers for authentication purposes.