Disable suppress_logs_and_warning in cli when debugging (#13180)

* Disable suppress_logs_and_warning in cli when debugging

In some cases commands like 'dags list' can be used for debug purposes.
The problem is that we are suppresing logs and warnings in some cases
to make the output nice and clean. This commit disable this functionality
if logging level is set to DEBUG.

* Add verbose flag

* fixup! Add verbose flag

airflow/cli/cli_parser.py

@@ -721,13 +721,13 @@ class GroupCommand(NamedTuple):
         name='list',
         help="List all the DAGs",
         func=lazy_load_command('airflow.cli.commands.dag_command.dag_list_dags'),
-        args=(ARG_SUBDIR, ARG_OUTPUT),
+        args=(ARG_SUBDIR, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='report',
         help='Show DagBag loading report',
         func=lazy_load_command('airflow.cli.commands.dag_command.dag_report'),
-        args=(ARG_SUBDIR, ARG_OUTPUT),
+        args=(ARG_SUBDIR, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='list-runs',
@@ -740,13 +740,21 @@ class GroupCommand(NamedTuple):
             "all the dagruns that were executed after this date. "
         ),
         func=lazy_load_command('airflow.cli.commands.dag_command.dag_list_dag_runs'),
-        args=(ARG_DAG_ID_OPT, ARG_NO_BACKFILL, ARG_STATE, ARG_OUTPUT, ARG_START_DATE, ARG_END_DATE),
+        args=(
+            ARG_DAG_ID_OPT,
+            ARG_NO_BACKFILL,
+            ARG_STATE,
+            ARG_OUTPUT,
+            ARG_VERBOSE,
+            ARG_START_DATE,
+            ARG_END_DATE,
+        ),
     ),
     ActionCommand(
         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, ARG_VERBOSE),
     ),
     ActionCommand(
         name='state',
@@ -980,50 +988,39 @@ class GroupCommand(NamedTuple):
         name='states-for-dag-run',
         help="Get the status of all task instances in a dag run",
         func=lazy_load_command('airflow.cli.commands.task_command.task_states_for_dag_run'),
-        args=(ARG_DAG_ID, ARG_EXECUTION_DATE, ARG_OUTPUT),
+        args=(ARG_DAG_ID, ARG_EXECUTION_DATE, ARG_OUTPUT, ARG_VERBOSE),
     ),
 )
 POOLS_COMMANDS = (
     ActionCommand(
         name='list',
         help='List pools',
         func=lazy_load_command('airflow.cli.commands.pool_command.pool_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='get',
         help='Get pool size',
         func=lazy_load_command('airflow.cli.commands.pool_command.pool_get'),
-        args=(
-            ARG_POOL_NAME,
-            ARG_OUTPUT,
-        ),
+        args=(ARG_POOL_NAME, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='set',
         help='Configure pool',
         func=lazy_load_command('airflow.cli.commands.pool_command.pool_set'),
-        args=(
-            ARG_POOL_NAME,
-            ARG_POOL_SLOTS,
-            ARG_POOL_DESCRIPTION,
-            ARG_OUTPUT,
-        ),
+        args=(ARG_POOL_NAME, ARG_POOL_SLOTS, ARG_POOL_DESCRIPTION, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='delete',
         help='Delete pool',
         func=lazy_load_command('airflow.cli.commands.pool_command.pool_delete'),
-        args=(
-            ARG_POOL_NAME,
-            ARG_OUTPUT,
-        ),
+        args=(ARG_POOL_NAME, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='import',
         help='Import pools',
         func=lazy_load_command('airflow.cli.commands.pool_command.pool_import'),
-        args=(ARG_POOL_IMPORT,),
+        args=(ARG_POOL_IMPORT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='export',
@@ -1037,13 +1034,13 @@ class GroupCommand(NamedTuple):
         name='list',
         help='List variables',
         func=lazy_load_command('airflow.cli.commands.variable_command.variables_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='get',
         help='Get variable',
         func=lazy_load_command('airflow.cli.commands.variable_command.variables_get'),
-        args=(ARG_VAR, ARG_JSON, ARG_DEFAULT),
+        args=(ARG_VAR, ARG_JSON, ARG_DEFAULT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='set',
@@ -1114,13 +1111,13 @@ class GroupCommand(NamedTuple):
         name='get',
         help='Get a connection',
         func=lazy_load_command('airflow.cli.commands.connection_command.connections_get'),
-        args=(ARG_CONN_ID, ARG_COLOR, ARG_OUTPUT),
+        args=(ARG_CONN_ID, ARG_COLOR, ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='list',
         help='List connections',
         func=lazy_load_command('airflow.cli.commands.connection_command.connections_list'),
-        args=(ARG_OUTPUT, ARG_CONN_ID_FILTER),
+        args=(ARG_OUTPUT, ARG_VERBOSE, ARG_CONN_ID_FILTER),
     ),
     ActionCommand(
         name='add',
@@ -1161,37 +1158,40 @@ class GroupCommand(NamedTuple):
         name='list',
         help='List installed providers',
         func=lazy_load_command('airflow.cli.commands.provider_command.providers_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='get',
         help='Get detailed information about a provider',
         func=lazy_load_command('airflow.cli.commands.provider_command.provider_get'),
-        args=(ARG_OUTPUT, ARG_FULL, ARG_COLOR, ARG_PROVIDER_NAME),
+        args=(ARG_OUTPUT, ARG_VERBOSE, ARG_FULL, ARG_COLOR, ARG_PROVIDER_NAME),
     ),
     ActionCommand(
         name='links',
         help='List extra links registered by the providers',
         func=lazy_load_command('airflow.cli.commands.provider_command.extra_links_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='widgets',
         help='Get information about registered connection form widgets',
         func=lazy_load_command('airflow.cli.commands.provider_command.connection_form_widget_list'),
-        args=(ARG_OUTPUT,),
+        args=(
+            ARG_OUTPUT,
+            ARG_VERBOSE,
+        ),
     ),
     ActionCommand(
         name='hooks',
         help='List registered provider hooks',
         func=lazy_load_command('airflow.cli.commands.provider_command.hooks_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='behaviours',
         help='Get information about registered connection types with custom behaviours',
         func=lazy_load_command('airflow.cli.commands.provider_command.connection_field_behaviours'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
 )
 
@@ -1200,7 +1200,7 @@ class GroupCommand(NamedTuple):
         name='list',
         help='List users',
         func=lazy_load_command('airflow.cli.commands.user_command.users_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='create',
@@ -1263,13 +1263,13 @@ class GroupCommand(NamedTuple):
         name='list',
         help='List roles',
         func=lazy_load_command('airflow.cli.commands.role_command.roles_list'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     ActionCommand(
         name='create',
         help='Create role',
         func=lazy_load_command('airflow.cli.commands.role_command.roles_create'),
-        args=(ARG_ROLES,),
+        args=(ARG_ROLES, ARG_VERBOSE),
     ),
 )
 
@@ -1443,7 +1443,7 @@ class GroupCommand(NamedTuple):
         name='cheat-sheet',
         help="Display cheat sheet",
         func=lazy_load_command('airflow.cli.commands.cheat_sheet_command.cheat_sheet'),
-        args=(),
+        args=(ARG_VERBOSE,),
     ),
     GroupCommand(
         name='connections',
@@ -1490,13 +1490,14 @@ class GroupCommand(NamedTuple):
         args=(
             ARG_ANONYMIZE,
             ARG_FILE_IO,
+            ARG_VERBOSE,
         ),
     ),
     ActionCommand(
         name='plugins',
         help='Dump information about loaded plugins',
         func=lazy_load_command('airflow.cli.commands.plugins_command.dump_plugins'),
-        args=(ARG_OUTPUT,),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
     GroupCommand(
         name="celery",

airflow/cli/commands/cheat_sheet_command.py

@@ -25,7 +25,7 @@
 from airflow.utils.helpers import partition
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def cheat_sheet(args):
     """Display cheat-sheet."""
     display_commands_index()

airflow/cli/commands/connection_command.py

@@ -52,7 +52,7 @@ def _connection_mapper(conn: Connection) -> Dict[str, Any]:
     }
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def connections_get(args):
     """Get a connection."""
     try:
@@ -66,7 +66,7 @@ def connections_get(args):
     )
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def connections_list(args):
     """Lists all connections at the command line"""
     with create_session() as session:

airflow/cli/commands/dag_command.py

@@ -269,7 +269,7 @@ def dag_next_execution(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def dag_list_dags(args):
     """Displays dags with or without stats at the command line"""
     dagbag = DagBag(process_subdir(args.subdir))
@@ -286,7 +286,7 @@ def dag_list_dags(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def dag_report(args):
     """Displays dagbag stats at the command line"""
     dagbag = DagBag(process_subdir(args.subdir))
@@ -304,7 +304,7 @@ def dag_report(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def dag_list_jobs(args, dag=None):
     """Lists latest n jobs"""
     queries = []
@@ -339,7 +339,7 @@ def dag_list_jobs(args, dag=None):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def dag_list_dag_runs(args, dag=None):
     """Lists dag runs for a given DAG"""
     if dag:

airflow/cli/commands/info_command.py

@@ -413,7 +413,7 @@ def _send_report_to_fileio(info):
         print(str(ex))
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def show_info(args):
     """Show information related to Airflow, system and other."""
     # Enforce anonymization, when file_io upload is tuned on.

airflow/cli/commands/plugins_command.py

@@ -49,7 +49,7 @@ def _join_plugins_names(value: Union[List[Any], Any]) -> str:
     return ",".join(_get_name(v) for v in value)
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def dump_plugins(args):
     """Dump plugins information"""
     plugins_manager.ensure_plugins_loaded()

airflow/cli/commands/pool_command.py

@@ -39,15 +39,15 @@ def _show_pools(pools, output):
     )
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def pool_list(args):
     """Displays info of all the pools"""
     api_client = get_current_api_client()
     pools = api_client.get_pools()
     _show_pools(pools=pools, output=args.output)
 
 
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def pool_get(args):
     """Displays pool info by a given name"""
     api_client = get_current_api_client()
@@ -59,7 +59,7 @@ def pool_get(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def pool_set(args):
     """Creates new pool with a given name and slots"""
     api_client = get_current_api_client()
@@ -68,7 +68,7 @@ def pool_set(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def pool_delete(args):
     """Deletes pool by a given name"""
     api_client = get_current_api_client()
@@ -80,7 +80,7 @@ def pool_delete(args):
 
 
 @cli_utils.action_logging
-@suppress_logs_and_warning()
+@suppress_logs_and_warning
 def pool_import(args):
     """Imports pools from the file"""
     if not os.path.exists(args.file):