def generate_backfill_command()

in utils/backfill.py [0:0]

• 21 lines of code
• 6 McCabe index (conditional complexity)

    def generate_backfill_command(self) -> list[str]:
        """
        Backfill command based off the Airflow plugin implemented by hwoo.

        Original implementation in plugins/backfill/main.py

        """
        # Construct the airflow command
        cmd = ["airflow"]

        if self.clear:
            cmd.extend(["tasks", "clear"])

            if self.dry_run:
                # For dry runs we simply time out to avoid zombie procs waiting on user input.
                # The output is what we're interested in
                timeout_list = ["timeout", "60"]
                cmd = timeout_list + cmd
            else:
                cmd.append("-y")

            if self.task_regex:
                cmd.extend(["-t", str(self.task_regex)])
        else:
            cmd.extend(["dags", "backfill", "--donot-pickle"])
            if self.dry_run:
                cmd.append("--dry-run")

            if self.task_regex:
                cmd.extend(["-t", str(self.task_regex)])

        cmd.extend(
            ["-s", str(self.start_date), "-e", str(self.end_date), str(self.dag_name)]
        )

        return cmd