This section provides an example of how you can use the Python collection and processing agents for a batch workflow. In this instance the Python collection agent reads an Excel file.

Workflow example with the Python collection and processing agents for batch

The Excel file being processed

Workflow Configuration

The sections below provide descriptions of agent configurations for this example workflow.

Workflow configuration, showing the workflow table

Read_Excel

The Python collection agent is configured to read all the Excel files from the configured directory, process each file in a separate batch and output one UDR per row.

General Tab

Example - Code for Read_Excel

import os
import pandas
from .UFL_Types import ItemCount

txn_directory = persistent()
txn_filename = persistent()

def execute():
    # Read all excel files from the configured directory
    txn_directory.value = dynamicFieldGet("Directory", "Name")
    xls = [f for f in os.listdir(txn_directory.value) if f.endswith(".xlsx")]
    # Process each excel file in a separate batch, output one UDR per row
    for fn in xls:
        if not isStopped():
            txn_filename.value = fn
            df = pandas.read_excel(os.path.join(txn_directory.value, fn))
            mimSet("Filename", fn)
            beginBatch()
            debug("Begin batch (txnid=%s, filename=%r)" % (getTransactionId(), fn))
            for idx, item, count in df.itertuples():
                udr = ItemCount()
                udr.item = item
                udr.count = count
                udrRoute(udr)
            endBatch()

def commit():
    # On commit, move the processed file into the "done/" subdirectory
    debug("Commit (txnid=%s, filename=%r)" % (getTransactionId(), txn_filename.value))
    old = os.path.join(txn_directory.value, txn_filename.value)
    new = os.path.join(os.path.join(txn_directory.value, "done"), txn_filename.value)
    if os.path.exists(old):
        os.rename(old, new)

MIM Tab

The agent publishes Filename as a string.

Analyse_Rows

The Python processing agent processes UDRs by defining a consume block, and uses the selected Interpreter profile that is used to configure the Python executables.

Example - Code for Analyse_Rows

total_count = 0

def beginBatch():
    fn = mimGet("Read_Excel", "Filename")
    debug("Begin batch (txnid=%s, filename=%r)" % (getTransactionId(), fn))
    global total_count
    total_count = 0

def consume(input):
    debug("%s=%s" % (input.item, input.count))
    global total_count
    total_count += input.count

def endBatch():
    fn = mimGet("Read_Excel", "Filename")
    debug("File %r contained %s items" % (fn, total_count))

Workflow Monitor

The output is shown step by step in the workflow monitor when you have debug enabled.

Next:

9.63.8 Listing Python Processes

MediationZone Documentation 8.3

9.63.7.3 Python Agents Example - Batch