Clean up command-line arguments (#11682)

* Clean up command-line arguments

- The `--do-pickle` argument is exclusively used by the scheduler, the `worker` command completely ignores it.
- Remove the `ARG_DAG_ID_OPT` entry entirely. The scheduler doesn't take a `--dag-id` option, and while the `dags list-runs` and `dags list-jobs` commands *do* take such a switch, the help text here is incorrect, they simply need to use `ARG_DAG_ID`.

* Put ARG_DAG_ID_OPT back

It differs from ARG_DAG_ID in that it is an optional command-line switch. I changed the help text to match its use: to select a specific DAG when listing runs or jobs. The dag is not actually triggered.

airflow/cli/cli_parser.py

@@ -201,6 +201,10 @@ def positive_int(value):
     default=ColorMode.AUTO)
 
 # list_dag_runs
+ARG_DAG_ID_OPT = Arg(
+    ("-d", "--dag-id"),
+    help="The id of the dag"
+)
 ARG_NO_BACKFILL = Arg(
     ("--no-backfill",),
     help="filter all the backfill dagruns given the dag id",
@@ -523,17 +527,11 @@ def positive_int(value):
          "stderr")
 
 # scheduler
-ARG_DAG_ID_OPT = Arg(
-    ("-d", "--dag-id"),
-    help="The id of the dag to run"
-)
 ARG_NUM_RUNS = Arg(
     ("-n", "--num-runs"),
     default=conf.getint('scheduler', 'num_runs'),
     type=int,
     help="Set the number of runs to execute before exiting")
-
-# worker
 ARG_DO_PICKLE = Arg(
     ("-p", "--do-pickle"),
     default=False,
@@ -542,6 +540,8 @@ def positive_int(value):
         "to the workers, instead of letting workers run their version "
         "of the code"),
     action="store_true")
+
+# worker
 ARG_QUEUES = Arg(
     ("-q", "--queues"),
     help="Comma delimited list of queues to serve",
@@ -820,7 +820,7 @@ class GroupCommand(NamedTuple):
         name='list-jobs',
         help="List the jobs",
         func=lazy_load_command('airflow.cli.commands.dag_command.dag_list_jobs'),
-        args=(ARG_DAG_ID_OPT, ARG_STATE, ARG_LIMIT, ARG_OUTPUT,),
+        args=(ARG_DAG_ID_OPT, ARG_STATE, ARG_LIMIT, ARG_OUTPUT),
     ),
     ActionCommand(
         name='state',
@@ -1236,7 +1236,7 @@ class GroupCommand(NamedTuple):
         help="Start a Celery worker node",
         func=lazy_load_command('airflow.cli.commands.celery_command.worker'),
         args=(
-            ARG_DO_PICKLE, ARG_QUEUES, ARG_CONCURRENCY, ARG_CELERY_HOSTNAME, ARG_PID, ARG_DAEMON,
+            ARG_QUEUES, ARG_CONCURRENCY, ARG_CELERY_HOSTNAME, ARG_PID, ARG_DAEMON,
             ARG_UMASK, ARG_STDOUT, ARG_STDERR, ARG_LOG_FILE, ARG_AUTOSCALE, ARG_SKIP_SERVE_LOGS
         ),
     ),