def process()

in runinferenceutil/infra.py [0:0]

• 7 lines of code
• 2 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"ParDoMerge window_start: {window_start}, window_end: {window_end}")

        for i in side:
            # print(f"Main {e.decode('utf-8')} Side {i}")
            # print(f"the side input extracted text {i.get('prompt')}")
            yield i.get('prompt') + '"' + element.decode('utf-8') + '"'