Remove unnecessary comprehensions

airflow/executors/celery_executor.py

@@ -260,7 +260,7 @@ def update_task_state(self, key: TaskInstanceKey, state: str, info: Any) -> None
 
     def end(self, synchronous: bool = False) -> None:
         if synchronous:
-            while any([task.state not in celery_states.READY_STATES for task in self.tasks.values()]):
+            while any(task.state not in celery_states.READY_STATES for task in self.tasks.values()):
                 time.sleep(5)
         self.sync()
 

airflow/models/baseoperator.py

@@ -961,7 +961,7 @@ def resolve_template_files(self) -> None:
                 if content is None:  # pylint: disable=no-else-continue
                     continue
                 elif isinstance(content, str) and \
-                        any([content.endswith(ext) for ext in self.template_ext]):
+                        any(content.endswith(ext) for ext in self.template_ext):
                     env = self.get_template_env()
                     try:
                         setattr(self, field, env.loader.get_source(env, content)[0])
@@ -971,7 +971,7 @@ def resolve_template_files(self) -> None:
                     env = self.dag.get_template_env()
                     for i in range(len(content)):  # pylint: disable=consider-using-enumerate
                         if isinstance(content[i], str) and \
-                                any([content[i].endswith(ext) for ext in self.template_ext]):
+                                any(content[i].endswith(ext) for ext in self.template_ext):
                             try:
                                 content[i] = env.loader.get_source(env, content[i])[0]
                             except Exception as e:  # pylint: disable=broad-except

airflow/operators/sql.py

@@ -95,7 +95,7 @@ def execute(self, context=None):
         self.log.info("Record: %s", records)
         if not records:
             raise AirflowException("The query returned None")
-        elif not all([bool(r) for r in records]):
+        elif not all(bool(r) for r in records):
             raise AirflowException(
                 "Test failed.\nQuery:\n{query}\nResults:\n{records!s}".format(
                     query=self.sql, records=records

airflow/utils/file.py

@@ -207,4 +207,4 @@ def might_contain_dag(file_path: str, safe_mode: bool, zip_file: Optional[zipfil
             return True
         with open(file_path, 'rb') as dag_file:
             content = dag_file.read()
-    return all([s in content for s in (b'DAG', b'airflow')])
+    return all(s in content for s in (b'DAG', b'airflow'))

airflow/www/security.py

@@ -323,7 +323,7 @@ def _has_role(self, role_name_or_list):
         if not isinstance(role_name_or_list, list):
             role_name_or_list = [role_name_or_list]
         return any(
-            [r.name in role_name_or_list for r in self.get_user_roles()])
+            r.name in role_name_or_list for r in self.get_user_roles())
 
     def _has_perm(self, permission_name, view_menu_name):
         """

backport_packages/import_all_provider_classes.py

@@ -46,8 +46,8 @@ def import_all_provider_classes(source_path: str,
     imported_classes = []
     tracebacks = []
     for root, _, files in os.walk(source_path):
-        if all([not root.startswith(prefix_provider_path)
-                for prefix_provider_path in prefixed_provider_paths]) or root.endswith("__pycache__"):
+        if all(not root.startswith(prefix_provider_path)
+                for prefix_provider_path in prefixed_provider_paths) or root.endswith("__pycache__"):
             # Skip loading module if it is not in the list of providers that we are looking for
             continue
         package_name = root[len(source_path) + 1:].replace("/", ".")

tests/jobs/test_backfill_job.py

@@ -910,7 +910,7 @@ def test_backfill_max_limit_check_within_limit(self):
 
         dagruns = DagRun.find(dag_id=dag.dag_id)
         self.assertEqual(2, len(dagruns))
-        self.assertTrue(all([run.state == State.SUCCESS for run in dagruns]))
+        self.assertTrue(all(run.state == State.SUCCESS for run in dagruns))
 
     def test_backfill_max_limit_check(self):
         dag_id = 'test_backfill_max_limit_check'
@@ -1437,7 +1437,7 @@ def test_backfill_run_backwards(self):
 
         queued_times = [ti.queued_dttm for ti in tis]
         self.assertTrue(queued_times == sorted(queued_times, reverse=True))
-        self.assertTrue(all([ti.state == State.SUCCESS for ti in tis]))
+        self.assertTrue(all(ti.state == State.SUCCESS for ti in tis))
 
         dag.clear()
         session.close()

tests/models/test_taskinstance.py

@@ -862,7 +862,7 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
             done=done,
             flag_upstream_failed=flag_upstream_failed,
         )
-        completed = all([dep.passed for dep in dep_results])
+        completed = all(dep.passed for dep in dep_results)
 
         self.assertEqual(completed, expect_completed)
         self.assertEqual(ti.state, expect_state)