Purpose and Use Case
Python Code Workers are one of the most flexible workers for customising a zSYNC pipeline. They have a single configuration field which accepts zSYNC code and allows importing of standard Python libraries, execution of standard Python code, as well as access to the pipeline data passed to the worker, and the instance environment.
Examples
1. Change the processed data attribute
data.data record.record2. Apply a filter
#reduce to only "updated"
updated = [x[1] for x in data.updated]
updated3. Convert date formats
from datetime import datetime
for d in data:
# Convert to datetime object
date_obj = datetime.strptime(d['date'], '%Y.%m.%d')
# Reformat date and update data
d['date'] = date_obj.strftime('%d/%m/%Y')
data4. Browse Odoo environment data
updated = record.updated[0][0]
env['sale.order'].browse(updated).action_confirm()
updated5. Transform data
from itertools import groupby
from typing import List, Dict, Any
def extract_order_number(item: Any) -> str:
item = item._d # Access the internal dictionary attribute
if isinstance(item, dict):
return item.get('Order Number', 'Error')
return "Error"
def transform_data(dataSet: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
# Sort data based according to order number
sorted_data = sorted(dataSet, key=extract_order_number)
result = []
for order_number, grouped_items in groupby(sorted_data, extract_order_number):
grouped_list = list(grouped_items)
# Skip any group with an error
if order_number == "Error":
continue
# Create a structured order dictionary
order = {
'Order Number': order_number,
'Date': grouped_list[0]['Date'],
'Customer': {
'Country': grouped_list[0]['Billing Country'],
'Region': grouped_list[0]['Billing Region'],
'City': grouped_list[0]['Billing City']
},
'Shipping Address': {
'Country': grouped_list[0]['Shipping Country'],
'Region': grouped_list[0]['Shipping Region'],
'City': grouped_list[0]['Shipping City']
},
'Lines': grouped_list
}
result.append(order)
return result
return transform_data(data)