def get_latest_update_time()

in infra-as-code/modules/export-to-bq-incremental/function-source-code/lib.py [0:0]

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

    def get_latest_update_time(self):
        query = f'''
            SELECT MAX(conversationUpdateTimestampUtc) as maxUpdateTimestamp FROM `{self.final_table_id}`
        '''

        bq_job = self.bq_client.query(query)

        bq_job_result = bq_job.result()

        maxUpdateTimestamp = None
        for row in bq_job_result:
            maxUpdateTimestamp = row['maxUpdateTimestamp']

        if maxUpdateTimestamp is not None:
            print(f'maxUpdateTimestamp found: `{maxUpdateTimestamp}`')

            dt = datetime.datetime.fromtimestamp(maxUpdateTimestamp)
            formatted_time = dt.strftime('%Y-%m-%dT%H:%M:%SZ')

            print(f'maxUpdateTimestamp found (formatted): `{formatted_time}`')

            return formatted_time
        else:
            print(f'maxUpdateTimestamp not found!')
            return None