def process()

in runinferenceutil/infra.py [0:0]

• 17 lines of code
• 3 McCabe index (conditional complexity)

    def process(self, element, side,  window=beam.DoFn.WindowParam):

        ts_format = '%Y-%m-%d %H:%M:%S.%f UTC'
        window_start = window.start.to_utc_datetime().strftime(ts_format)
        window_end = window.end.to_utc_datetime().strftime(ts_format)
        logging.info(f"FormatForBigquery window_start: {window_start}, window_end: {window_end}")

        for i in side:
            now = datetime.now()  # current date and time
            date_time = now.strftime("%Y-%m-%d %H:%M:%S")

            item = element['Yes']
            percent, total = element['Yes'][0], element['Yes'][1]

            prompt = i.get('prompt')

            return [{
                'time': date_time,
                'prompt' : prompt,
                'totalMessages': total,
                'detectionPercent': percent,
            }]