def _take_action()

in tools/cloud-build/babysit/core.py [0:0]

• 20 lines of code
• 13 McCabe index (conditional complexity)

    def _take_action(self, builds: Sequence[Build]) -> bool:
        """
        Returns bool - whether "babysitting" should be continued.
        """
        latest = latest_by_trigger(builds).values()
        active = [bc.build for bc in latest if not self._in_terminal_state(bc)]
        if not active:
            return False  # all builds are in terminal state, done

        not_running = [b for b in active if b.status not in (Status.QUEUED, Status.WORKING)]
        num_running = len(active) - len(not_running)

        if num_running == len(active):
            return True  # waiting for results
        if num_running >= self.concurrency:
            return True  # waiting for "opening"

        pending = [b for b in not_running if b.status == Status.PENDING]
        if pending:  # approve one of pending builds
            self._approve(random.choice(pending))
            return True

        assert not_running  # sanity check
        failed = random.choice(not_running)
        assert failed.status in [Status.FAILURE, Status.INTERNAL_ERROR, Status.TIMEOUT]  # sanity check
        self._retry(failed)  # retry failed build
        return True