remove superfluous else block

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-db00d57fce32830b69f2c1481b231e65e67e197b4a96a5fa1c870cd555eac3bd
+7d6f2aa31fb10d8006b6b7f572bedcc4be78eb828edfd42386e0b872b6999afc

airflow-core/src/airflow/__init__.py

@@ -101,7 +101,7 @@ def __getattr__(name: str):
     module_path, attr_name, deprecated = __lazy_imports.get(name, ("", "", False))
     if not module_path:
         raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-    elif deprecated:
+    if deprecated:
         warnings.warn(
             f"Import {name!r} directly from the airflow module is deprecated and "
             f"will be removed in the future. Please import it from 'airflow{module_path}.{attr_name}'.",

airflow-core/src/airflow/api_fastapi/auth/tokens.py

@@ -92,10 +92,9 @@ def _guess_best_algorithm(key: AllowedPrivateKeys):
 
     if isinstance(key, RSAPrivateKey):
         return "RS512"
-    elif isinstance(key, Ed25519PrivateKey):
+    if isinstance(key, Ed25519PrivateKey):
         return "EdDSA"
-    else:
-        raise ValueError(f"Unknown key object {type(key)}")
+    raise ValueError(f"Unknown key object {type(key)}")
 
 
 @attrs.define(repr=False)
@@ -297,8 +296,7 @@ def __attrs_post_init__(self):
                     "Cannot guess the algorithm when using JWKS - please specify it in the config option "
                     "[api_auth] jwt_algorithm"
                 )
-            else:
-                self.algorithm = ["HS512"]
+            self.algorithm = ["HS512"]
 
     def _get_kid_from_header(self, unvalidated: str) -> str:
         header = jwt.get_unverified_header(unvalidated)
@@ -475,7 +473,7 @@ def generate_private_key(key_type: str = "RSA", key_size: int = 2048):
         # Generate an RSA private key
 
         return rsa.generate_private_key(public_exponent=65537, key_size=key_size, backend=default_backend())
-    elif key_type == "Ed25519":
+    if key_type == "Ed25519":
         return ed25519.Ed25519PrivateKey.generate()
     raise ValueError(f"unsupported key type: {key_type}")
 

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -206,8 +206,7 @@ def to_orm(self, select: Select) -> Select:
 
         if self.value[0] == "-":
             return select.order_by(nullscheck, column.desc(), primary_key_column.desc())
-        else:
-            return select.order_by(nullscheck, column.asc(), primary_key_column.asc())
+        return select.order_by(nullscheck, column.asc(), primary_key_column.asc())
 
     def get_primary_key_column(self) -> Column:
         """Get the primary key column of the model of SortParam object."""

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py

@@ -88,7 +88,7 @@ def get_owners(cls, v: Any) -> list[str] | None:
 
         if v is None:
             return []
-        elif isinstance(v, str):
+        if isinstance(v, str):
             return v.split(",")
         return v
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -275,15 +275,14 @@ def clear_dag_run(
             task_instances=cast("list[TaskInstanceResponse]", task_instances),
             total_entries=len(task_instances),
         )
-    else:
-        dag.clear(
-            run_id=dag_run_id,
-            task_ids=None,
-            only_failed=body.only_failed,
-            session=session,
-        )
-        dag_run_cleared = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
-        return dag_run_cleared
+    dag.clear(
+        run_id=dag_run_id,
+        task_ids=None,
+        only_failed=body.only_failed,
+        session=session,
+    )
+    dag_run_cleared = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
+    return dag_run_cleared
 
 
 @dag_run_router.get(

airflow-core/src/airflow/api_fastapi/core_api/routes/public/log.py

@@ -140,16 +140,15 @@ def get_log(
         if not metadata.get("end_of_log", False):
             encoded_token = URLSafeSerializer(request.app.state.secret_key).dumps(metadata)
         return TaskInstancesLogResponse.model_construct(continuation_token=encoded_token, content=logs)
-    else:
-        # text/plain, or something else we don't understand. Return raw log content
-
-        # We need to exhaust the iterator before we can generate the continuation token.
-        # We could improve this by making it a streaming/async response, and by then setting the header using
-        # HTTP Trailers
-        logs = "".join(task_log_reader.read_log_stream(ti, try_number, metadata))
-        headers = None
-        if not metadata.get("end_of_log", False):
-            headers = {
-                "Airflow-Continuation-Token": URLSafeSerializer(request.app.state.secret_key).dumps(metadata)
-            }
-        return Response(media_type="application/x-ndjson", content=logs, headers=headers)
+    # text/plain, or something else we don't understand. Return raw log content
+
+    # We need to exhaust the iterator before we can generate the continuation token.
+    # We could improve this by making it a streaming/async response, and by then setting the header using
+    # HTTP Trailers
+    logs = "".join(task_log_reader.read_log_stream(ti, try_number, metadata))
+    headers = None
+    if not metadata.get("end_of_log", False):
+        headers = {
+            "Airflow-Continuation-Token": URLSafeSerializer(request.app.state.secret_key).dumps(metadata)
+        }
+    return Response(media_type="application/x-ndjson", content=logs, headers=headers)

airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py

@@ -97,7 +97,7 @@ def handle_bulk_create(
                     status_code=status.HTTP_409_CONFLICT,
                     detail=f"The connections with these connection_ids: {matched_connection_ids} already exist.",
                 )
-            elif action.action_on_existence == BulkActionOnExistence.SKIP:
+            if action.action_on_existence == BulkActionOnExistence.SKIP:
                 create_connection_ids = not_found_connection_ids
             else:
                 create_connection_ids = to_create_connection_ids
@@ -130,7 +130,7 @@ def handle_bulk_update(
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 update_connection_ids = matched_connection_ids
             else:
                 update_connection_ids = to_update_connection_ids
@@ -170,7 +170,7 @@ def handle_bulk_delete(
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 delete_connection_ids = matched_connection_ids
             else:
                 delete_connection_ids = to_delete_connection_ids

airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py

@@ -66,7 +66,7 @@ def handle_bulk_create(self, action: BulkCreateAction[PoolBody], results: BulkAc
                     status_code=status.HTTP_409_CONFLICT,
                     detail=f"The pools with these pool names: {matched_pool_names} already exist.",
                 )
-            elif action.action_on_existence == BulkActionOnExistence.SKIP:
+            if action.action_on_existence == BulkActionOnExistence.SKIP:
                 create_pool_names = not_found_pool_names
             else:
                 create_pool_names = to_create_pool_names
@@ -97,7 +97,7 @@ def handle_bulk_update(self, action: BulkUpdateAction[PoolBody], results: BulkAc
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 update_pool_names = matched_pool_names
             else:
                 update_pool_names = to_update_pool_names
@@ -134,7 +134,7 @@ def handle_bulk_delete(self, action: BulkDeleteAction[PoolBody], results: BulkAc
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 delete_pool_names = matched_pool_names
             else:
                 delete_pool_names = to_delete_pool_names

airflow-core/src/airflow/api_fastapi/core_api/services/public/variables.py

@@ -57,7 +57,7 @@ def handle_bulk_create(self, action: BulkCreateAction, results: BulkActionRespon
                     status_code=status.HTTP_409_CONFLICT,
                     detail=f"The variables with these keys: {matched_keys} already exist.",
                 )
-            elif action.action_on_existence == BulkActionOnExistence.SKIP:
+            if action.action_on_existence == BulkActionOnExistence.SKIP:
                 create_keys = not_found_keys
             else:
                 create_keys = to_create_keys
@@ -86,7 +86,7 @@ def handle_bulk_update(self, action: BulkUpdateAction, results: BulkActionRespon
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The variables with these keys: {not_found_keys} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 update_keys = matched_keys
             else:
                 update_keys = to_update_keys
@@ -118,7 +118,7 @@ def handle_bulk_delete(self, action: BulkDeleteAction, results: BulkActionRespon
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=f"The variables with these keys: {not_found_keys} were not found.",
                 )
-            elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+            if action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
                 delete_keys = matched_keys
             else:
                 delete_keys = to_delete_keys

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -195,13 +195,13 @@ def ti_state_discriminator(v: dict[str, str] | StrictBaseModel) -> str:
 
     if state == TIState.SUCCESS:
         return "success"
-    elif state in set(TerminalTIState):
+    if state in set(TerminalTIState):
         return "_terminal_"
-    elif state == TIState.DEFERRED:
+    if state == TIState.DEFERRED:
         return "deferred"
-    elif state == TIState.UP_FOR_RESCHEDULE:
+    if state == TIState.UP_FOR_RESCHEDULE:
         return "up_for_reschedule"
-    elif state == TIState.UP_FOR_RETRY:
+    if state == TIState.UP_FOR_RETRY:
         return "up_for_retry"
     return "_other_"
 

airflow-core/src/airflow/cli/commands/api_server_command.py

@@ -129,9 +129,9 @@ def _get_ssl_cert_and_key_filepaths(cli_arguments) -> tuple[str | None, str | No
             raise AirflowConfigException(error_template_2.format(ssl_key))
 
         return (ssl_cert, ssl_key)
-    elif ssl_cert:
+    if ssl_cert:
         raise AirflowConfigException(error_template_1.format("SSL certificate", "SSL key"))
-    elif ssl_key:
+    if ssl_key:
         raise AirflowConfigException(error_template_1.format("SSL key", "SSL certificate"))
 
     return (None, None)

airflow-core/src/airflow/cli/commands/dag_command.py

@@ -234,7 +234,7 @@ def dag_dependencies_show(args) -> None:
             "Option --save and --imgcat are mutually exclusive. "
             "Please remove one option to execute the command.",
         )
-    elif filename:
+    if filename:
         _save_dot_to_file(dot, filename)
     elif imgcat:
         _display_dot_via_imgcat(dot)
@@ -255,7 +255,7 @@ def dag_show(args) -> None:
             "Option --save and --imgcat are mutually exclusive. "
             "Please remove one option to execute the command.",
         )
-    elif filename:
+    if filename:
         _save_dot_to_file(dot, filename)
     elif imgcat:
         _display_dot_via_imgcat(dot)
@@ -275,8 +275,7 @@ def _display_dot_via_imgcat(dot: Dot) -> None:
     except OSError as e:
         if e.errno == errno.ENOENT:
             raise SystemExit("Failed to execute. Make sure the imgcat executables are on your systems 'PATH'")
-        else:
-            raise
+        raise
 
 
 def _save_dot_to_file(dot: Dot, filename: str) -> None:

airflow-core/src/airflow/cli/commands/info_command.py

@@ -141,11 +141,11 @@ def get_current() -> OperatingSystem:
         """Get current operating system."""
         if os.name == "nt":
             return OperatingSystem.WINDOWS
-        elif "linux" in sys.platform:
+        if "linux" in sys.platform:
             return OperatingSystem.LINUX
-        elif "darwin" in sys.platform:
+        if "darwin" in sys.platform:
             return OperatingSystem.MACOSX
-        elif "cygwin" in sys.platform:
+        if "cygwin" in sys.platform:
             return OperatingSystem.CYGWIN
         return OperatingSystem.UNKNOWN
 
@@ -203,8 +203,7 @@ def _get_version(cmd: list[str], grep: bytes | None = None):
                     data = [line for line in data if grep in line]
                 if len(data) != 1:
                     return "NOT AVAILABLE"
-                else:
-                    return data[0].decode()
+                return data[0].decode()
         except OSError:
             return "NOT AVAILABLE"
 
@@ -216,8 +215,7 @@ def get_fullname(o):
             module = o.__class__.__module__
             if module is None or module == str.__class__.__module__:
                 return o.__class__.__name__  # Avoid reporting __builtin__
-            else:
-                return f"{module}.{o.__class__.__name__}"
+            return f"{module}.{o.__class__.__name__}"
 
         try:
             handler_names = [get_fullname(handler) for handler in logging.getLogger("airflow.task").handlers]

airflow-core/src/airflow/cli/commands/task_command.py

@@ -108,7 +108,7 @@ def _get_dag_run(
         )
         if dag_run is not None:
             return dag_run, False
-        elif not create_if_necessary:
+        if not create_if_necessary:
             raise DagRunNotFound(
                 f"DagRun for {dag.dag_id} with run_id or logical_date of {logical_date_or_run_id!r} not found"
             )
@@ -132,7 +132,7 @@ def _get_dag_run(
             state=DagRunState.RUNNING,
         )
         return dag_run, True
-    elif create_if_necessary == "db":
+    if create_if_necessary == "db":
         dag_run = dag.create_dagrun(
             run_id=_generate_temporary_run_id(),
             logical_date=dag_run_logical_date,

airflow-core/src/airflow/configuration.py

@@ -124,8 +124,7 @@ def expand_env_var(env_var: str | None) -> str | None:
         interpolated = os.path.expanduser(os.path.expandvars(str(env_var)))
         if interpolated == env_var:
             return interpolated
-        else:
-            env_var = interpolated
+        env_var = interpolated
 
 
 def run_command(command: str) -> str:
@@ -1160,13 +1159,12 @@ def getboolean(self, section: str, key: str, **kwargs) -> bool:  # type: ignore[
             val = val.split("#")[0].strip()
         if val in ("t", "true", "1"):
             return True
-        elif val in ("f", "false", "0"):
+        if val in ("f", "false", "0"):
             return False
-        else:
-            raise AirflowConfigException(
-                f'Failed to convert value to bool. Please check "{key}" key in "{section}" section. '
-                f'Current value: "{val}".'
-            )
+        raise AirflowConfigException(
+            f'Failed to convert value to bool. Please check "{key}" key in "{section}" section. '
+            f'Current value: "{val}".'
+        )
 
     def getint(self, section: str, key: str, **kwargs) -> int:  # type: ignore[override]
         val = self.get(section, key, _extra_stacklevel=1, **kwargs)
@@ -2020,7 +2018,7 @@ def write_default_airflow_configuration_if_needed() -> AirflowConfigParser:
             f"but got a directory {airflow_config.__fspath__()!r}."
         )
         raise IsADirectoryError(msg)
-    elif not airflow_config.exists():
+    if not airflow_config.exists():
         log.debug("Creating new Airflow config file in: %s", airflow_config.__fspath__())
         config_directory = airflow_config.parent
         if not config_directory.exists():

airflow-core/src/airflow/dag_processing/bundles/base.py

@@ -48,8 +48,7 @@
 def get_bundle_storage_root_path():
     if configured_location := conf.get("dag_processor", "dag_bundle_storage_path", fallback=None):
         return Path(configured_location)
-    else:
-        return Path(tempfile.gettempdir(), "airflow", "dag_bundles")
+    return Path(tempfile.gettempdir(), "airflow", "dag_bundles")
 
 
 STALE_BUNDLE_TRACKING_FOLDER = get_bundle_storage_root_path() / "_tracking"

airflow-core/src/airflow/dag_processing/processor.py

@@ -147,7 +147,7 @@ def _execute_callbacks(
                 "Haven't coded Task callback yet - https://github.com/apache/airflow/issues/44354!"
             )
             # _execute_task_callbacks(dagbag, request)
-        elif isinstance(request, DagCallbackRequest):
+        if isinstance(request, DagCallbackRequest):
             _execute_dag_callbacks(dagbag, request, log)
 
 
@@ -277,7 +277,7 @@ def _handle_request(self, msg: ToManager, log: FilteringBoundLogger) -> None:  #
         if isinstance(msg, DagFileParsingResult):
             self.parsing_result = msg
             return
-        elif isinstance(msg, GetConnection):
+        if isinstance(msg, GetConnection):
             conn = self.client.connections.get(msg.conn_id)
             if isinstance(conn, ConnectionResponse):
                 conn_result = ConnectionResult.from_conn_response(conn)

airflow-core/src/airflow/example_dags/example_bash_decorator.py

@@ -73,8 +73,7 @@ def this_will_skip() -> str:
     def sleep_in(day: str) -> str:
         if day in (WeekDay.SATURDAY, WeekDay.SUNDAY):
             return f"sleep {60 * 60}"
-        else:
-            raise AirflowSkipException("No sleeping in today!")
+        raise AirflowSkipException("No sleeping in today!")
 
     sleep_in(day="{{ dag_run.logical_date.strftime('%A').lower() }}")
     # [END howto_decorator_bash_conditional]

airflow-core/src/airflow/example_dags/example_branch_python_dop_operator_3.py

@@ -39,8 +39,7 @@ def should_run(**kwargs) -> str:
     print(f"------------- exec dttm = {kwargs['logical_date']} and minute = {kwargs['logical_date'].minute}")
     if kwargs["logical_date"].minute % 2 == 0:
         return "empty_task_1"
-    else:
-        return "empty_task_2"
+    return "empty_task_2"
 
 
 with DAG(