Refactor: Simplify comparisons (#34181)

airflow/dag_processing/manager.py

@@ -1081,7 +1081,7 @@ def start_new_processes(self):
         # needs to be done before this process is forked to create the DAG parsing processes.
         SecretCache.init()
 
-        while self._parallelism - len(self._processors) > 0 and self._file_path_queue:
+        while self._parallelism > len(self._processors) and self._file_path_queue:
             file_path = self._file_path_queue.popleft()
             # Stop creating duplicate processor i.e. processor with the same filepath
             if file_path in self._processors:

airflow/plugins_manager.py

@@ -340,9 +340,8 @@ def ensure_plugins_loaded():
         for plugin in plugins:
             registered_hooks.extend(plugin.hooks)
 
-    num_loaded = len(plugins)
-    if num_loaded > 0:
-        log.debug("Loading %d plugin(s) took %.2f seconds", num_loaded, timer.duration)
+    if plugins:
+        log.debug("Loading %d plugin(s) took %.2f seconds", len(plugins), timer.duration)
 
 
 def initialize_web_ui_plugins():

airflow/providers/amazon/aws/hooks/datasync.py

@@ -57,9 +57,10 @@ def __init__(self, wait_interval_seconds: int = 30, *args, **kwargs) -> None:
         self.locations: list = []
         self.tasks: list = []
         # wait_interval_seconds = 0 is used during unit tests
-        if wait_interval_seconds < 0 or wait_interval_seconds > 15 * 60:
+        if 0 <= wait_interval_seconds <= 15 * 60:
+            self.wait_interval_seconds = wait_interval_seconds
+        else:
             raise ValueError(f"Invalid wait_interval_seconds {wait_interval_seconds}")
-        self.wait_interval_seconds = wait_interval_seconds
 
     def create_location(self, location_uri: str, **create_location_kwargs) -> str:
         """

airflow/providers/apache/hive/hooks/hive.py

@@ -172,7 +172,7 @@ def _validate_beeline_parameters(self, conn):
             )
         try:
             int_port = int(conn.port)
-            if int_port <= 0 or int_port > 65535:
+            if not 0 < int_port <= 65535:
                 raise Exception(f"The port used in beeline command ({conn.port}) should be in range 0-65535)")
         except (ValueError, TypeError) as e:
             raise Exception(f"The port used in beeline command ({conn.port}) should be a valid integer: {e})")

airflow/providers/google/cloud/operators/bigquery_dts.py

@@ -356,7 +356,7 @@ def execute(self, context: Context):
         )
 
     def _wait_for_transfer_to_be_done(self, run_id: str, transfer_config_id: str, interval: int = 10):
-        if interval < 0:
+        if interval <= 0:
             raise ValueError("Interval must be > 0")
 
         while True:

airflow/providers/google/cloud/sensors/dataproc_metastore.py

@@ -113,4 +113,4 @@ def poke(self, context: Context) -> bool:
 
         # Return True if we got all requested partitions.
         # If no partitions were given in the request, then we expect to find at least one.
-        return found_partitions > 0 and found_partitions >= len(set(self.partitions))
+        return found_partitions >= max(1, len(set(self.partitions)))

airflow/providers/google/cloud/transfers/gcs_to_gcs.py

@@ -307,7 +307,7 @@ def _ignore_existing_files(self, hook, prefix, **kwargs):
             ]
 
         objects = set(objects) - set(existing_objects)
-        if len(objects) > 0:
+        if objects:
             self.log.info("%s files are going to be synced: %s.", len(objects), objects)
         else:
             self.log.info("There are no new files to sync. Have a nice day!")

airflow/providers/snowflake/utils/sql_api_generate_jwt.py

@@ -102,14 +102,10 @@ def prepare_account_name_for_jwt(self, raw_account: str) -> str:
         account = raw_account
         if ".global" not in account:
             # Handle the general case.
-            idx = account.find(".")
-            if idx > 0:
-                account = account[0:idx]
+            account = account.partition(".")[0]
         else:
             # Handle the replication case.
-            idx = account.find("-")
-            if idx > 0:
-                account = account[0:idx]  # pragma: no cover
+            account = account.partition("-")[0]
         # Use uppercase for the account identifier.
         return account.upper()
 

tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -92,7 +92,7 @@ def _cases(self):
     @staticmethod
     def _is_valid_pod_id(name):
         regex = r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
-        return len(name) <= 253 and all(ch.lower() == ch for ch in name) and re.match(regex, name)
+        return len(name) <= 253 and name.islower() and re.match(regex, name)
 
     @staticmethod
     def _is_safe_label_value(value):