Apache Beam Pardo class for Exponea pipeline

pardo_class.py

class FetchEventsFn(beam.DoFn):

    def __init__(self, payload):
        self.payload = payload
        self.url = exponea_vars['url']
        self.headers = exponea_vars['headers']
    
    def process(self, record):
        try:
            id = record['cookie']
            payload = self.payload
            payload["customer_ids"]["cookie"] = id

            response = requests.request("POST", self.url, data=json.dumps(payload), headers=self.headers)

            response_json = response.json()

            l_properties = response_json.get("data")

            l_customer = [
                      {
                          "id": id, 
                          "events": [{k:str(v) for k,v in i.items()}]
                      } 
                      for i in l_properties
                  ] if l_properties else []
        
        except Exception as e:
            l_customer = []
            logging.error(f'{id} load failed due to {e}')

        return l_customer