def find_recent_partitions()

in athena_glue_service_logs/partitioners/date_partitioner.py [0:0]

• 14 lines of code
• 9 McCabe index (conditional complexity)

    def find_recent_partitions(self, existing_partitions):
        partitions_to_add = []

        # Now check to see if, in each region, that a partition day exists for today.
        # If it does not, backfill up to today.
        today = datetime.utcfromtimestamp(time.time()).date()
        day_diff = 0

        # Only go back MAX_RECENT_DAYS days for now and only if S3 objects actually exist...
        for _ in range(self.MAX_RECENT_DAYS):
            new_day = today + timedelta(days=day_diff)
            new_day_tuple = new_day.strftime('%Y-%m-%d').split('-')
            if not existing_partitions or existing_partitions[-1] != new_day_tuple:
                if S3Reader(self.build_partitioned_path(new_day_tuple)).does_have_objects():
                    partitions_to_add.append(new_day_tuple)
            else:
                break
            day_diff -= 1

        return partitions_to_add