# Full Price Flow

The Full Price flow ensures the synchronization of all product prices between ERP systems and Omnitron by fetching and comparing price data. This flow operates exclusively in an inbound mode and triggers itself once daily at a fixed time, which cannot be altered by the user and the flow can not be triggered manually. During each execution, it retrieves all prices from the ERP system, compares them to the existing prices in Omnitron, and identifies discrepancies. For products with price differences, the flow automatically triggers the corresponding price flow to fetch and update the individual product prices. By systematically identifying and updating mismatched prices, the FullPrice flow ensures accurate and consistent pricing data across the systems.

### <mark style="color:red;">Full Price Flow Types​</mark> <a href="#full-price-flow-types" id="full-price-flow-types"></a>

* **Inbound Flow:** Indicates that the flow will read data from the ERP system. The flow is exclusively inbound, it will always run once a day at the set time that can’t be altered by the user.

### <mark style="color:red;">Flow Steps​</mark> <a href="#flow-steps" id="flow-steps"></a>

* **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:** Price queries are made from the ERP system.
* **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 Mapping](jolt-mapping) documentation and [Jolt Transform](https://jolt-demo.appspot.com/#inception) website.
* **Compare Prices:** This step identifies discrepancies between the prices retrieved from the ERP system and those stored in Omnitron. It triggers the corresponding price flow for each product with a price difference to ensure accurate synchronization.

### <mark style="color:red;">Detailed Flow Designer Settings​</mark> <a href="#detailed-flow-designer-settings" id="detailed-flow-designer-settings"></a>

#### Configuration Card​ <a href="#configuration-card" id="configuration-card"></a>

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2Fq4l3cplJLaD6dvZ0gdgx%2Fimage.png?alt=media&#x26;token=17c8de6e-98be-4fd0-b883-fb42ef2a14df" alt=""><figcaption></figcaption></figure>

* **Omnitron Price List ID:**&#x49;nformation about the price list ID to be updated in Omnitron should be retrieved from Omnitron itself. To do this, navigate to Products and Catalogs, then Price List, and open the price list page. The flow is designed to update the price list ID, which can be found in the last digits of the URL. For example, in the URL [_https://demo.akinon.net/en/tr-tr/products-and-categories/prices/price-detail/1_](https://demo.akinon.net/en/tr-tr/products-and-categories/prices/price-detail/1), the price list ID is "1". This ID is also used in the "Fetch Missing" task to identify missing SKUs in the price list.
* **Omnitron API Settings:**
  * **Domain Url:** Omnitron's domain URL, **Example:** [https://demo.omnitron.akinon.net/](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`

      <figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2F7MeISEW1taueNPadBYVj%2Fimage.png?alt=media&#x26;token=aacbf614-31e4-4ed0-ac7d-89c362af5c38" alt=""><figcaption></figcaption></figure>

#### Read Data From ERP Card​ <a href="#read-data-from-erp-card" id="read-data-from-erp-card"></a>

*   **Add Extra Headers as Dict Format:** During the price query from ERP, additional headers to be sent.

    <figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2F8ueQFmaGA9RWVyh7HDOI%2Fimage.png?alt=media&#x26;token=f1fb4604-5a37-44b9-8987-2f2b2ca65d23" alt="" width="563"><figcaption></figcaption></figure>
* **Price List Key:** Key used as a parameter during queries from ERP.
* **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.
* **Endpoint URL:** URL for reading data from the ERP system.

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FoQI8Qgr5QHRvuABIHDMU%2Fimage.png?alt=media&#x26;token=c7075c47-0835-4dae-8e8a-718581bf4a26" alt="" width="563"><figcaption></figcaption></figure>

**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`

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

![](https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2F1e7KApfgH3kVetpMgHPN%2Fimage.png?alt=media\&token=b2f6b3e5-ba80-4c29-811a-aa9f71b969f4)
{% 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`&#x20;

    <figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FRQBW5FrtOh1yPPmdXfBp%2Fimage.png?alt=media&#x26;token=4ed37f68-6f7c-457d-a73a-a2c29ca71346" alt="" width="563"><figcaption></figcaption></figure>
* 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`

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2Fs4iJveUXyXCN1FD2SpeS%2Fimage.png?alt=media&#x26;token=9ca526d7-4313-408a-84aa-336dc3b7f39e" alt="" width="563"><figcaption></figcaption></figure>

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

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FyAbYwbRWAKj0klWEUorM%2Fimage.png?alt=media&#x26;token=1fafcc6c-e815-47be-bb57-82ac19583a26" alt="" width="563"><figcaption></figcaption></figure>

#### Script Card (Optional)​ <a href="#script-card-optional" id="script-card-optional"></a>

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 whose keys begin with the prefix `log.` will be automatically logged at the end of the execution if a log file is generated, ensuring important information is recorded and can be reviewed later.

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FflR4cC8JaOAB4BfgCosS%2Fimage.png?alt=media&#x26;token=b9cab139-8ea3-40b3-91c3-9ee3adc627eb" alt=""><figcaption></figcaption></figure>

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

For more information about script card, please refer to the [Script Document](scripting-in-flows).

#### Mapping Card (Optional)​ <a href="#mapping-card-optional" id="mapping-card-optional"></a>

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FEtPcnwbJdiMG7IinMVXh%2Fimage.png?alt=media&#x26;token=20f3a6af-eb7a-4847-84fa-f5f8425236e0" alt=""><figcaption></figcaption></figure>

Details can be found at the [Jolt Transform](https://jolt-demo.appspot.com/#inception) website. Because this is the last step the input data can be transformed, the flow expects the following output at the end of this step to work correctly.

**Example Mapping:**

```
[
    {
        "operation": "shift",
        "spec": {
            "*": {
                "sku": "[&1].product_sku",
                "price": "[&1].price",
                "price_list": "[&1].price_list",
                "currency_type": "[&1].currency_type",
                "tax_rate": "[&1].tax_rate",
                "retail_price": "[&1].retail_price"
            }
        }
    }
]
```

**Expected Output:**

```
[ {
  "product_sku" : "1234",
  "price" : 2211.0,
  "price_list" : 1,
  "currency_type" : "try",
  "tax_rate" : "0.00",
  "retail_price" : 2211.0
}, {
  "product_sku" : "12345",
  "price" : 1122.0,
  "price_list" : 1,
  "currency_type" : "try",
  "tax_rate" : "0.00",
  "retail_price" : 1122.0
}]
```

**Mapping Testing:** The response from the script is placed in the input field, and the result of the mapping is viewed in the result field.

#### Compare Prices​ <a href="#compare-prices" id="compare-prices"></a>

<figure><img src="https://3333414532-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FIbwGN7KwvYi0iLbjtnXz%2Fuploads%2FxPMGIxIsEaruYHT3dAPL%2Fimage.png?alt=media&#x26;token=c7842e6e-981d-4f46-8157-2305838eb88e" alt=""><figcaption></figcaption></figure>

**Price Flow:**

This is a dropdown field that is used for selecting related price flow for current Full Price flow. Using this dropdown it is possible to select one of the price flows of the project, for each product with a price difference this flow will be triggered to query the products from ERP one by one.

**Comparison Settings:**

This setting is used for comparing products between the ERP and Omnitron. An SQL like language is used for comparing the products, where a default query is provided when the flow is first created.

```
-- Define a Common Table Expression (CTE) called 'omnitron', which selects the 'sku' and 'price' columns from the 'FLOWFILE' table where the 'SOURCE' column equals 'omnitron'
WITH omnitron AS
  (SELECT sku,
        price
   FROM FLOWFILE
   WHERE SOURCE = 'omnitron'),

-- Define a CTE called 'erp', which selects the 'sku' and 'price' columns from the 'FLOWFILE' table where the 'SOURCE' column equals 'erp'
    erp AS
  (SELECT sku,
        price
   FROM FLOWFILE
   WHERE SOURCE = 'erp'),

-- Define a CTE called 'all_skus', which selects only the 'sku' column from the 'FLOWFILE' table
    all_skus AS
  (SELECT sku
   FROM FLOWFILE)

-- Select all 'sku' values from the 'all_skus' CTE, and left join with the 'omnitron' and 'erp' CTEs on their respective 'sku' columns
SELECT all_skus.sku AS sku
FROM all_skus
LEFT JOIN omnitron ON all_skus.sku = omnitron.sku
LEFT JOIN erp ON all_skus.sku = erp.sku

-- Filter the results based on the following conditions:
-- 1. If 'omnitron.price' is null and 'erp.price' is greater than 0, or
-- 2. If 'erp.price' is null and 'omnitron.price' is greater than 0, or
-- 3. If both 'omnitron.price' and 'erp.price' are not null, and they are not equal
WHERE (omnitron.price IS NULL
    AND erp.price > 0)
  OR (erp.price IS NULL
    AND omnitron.price > 0)
  OR (omnitron.price IS NOT NULL
    AND erp.price IS NOT NULL
    AND omnitron.price <> erp.price)

-- Group the results by the 'sku' column
GROUP BY all_skus.sku
```
