Replace to broad exceptions into the Core

airflow/api_connexion/schemas/common_schema.py

@@ -123,7 +123,7 @@ def get_obj_type(self, obj):
         elif isinstance(obj, CronExpression):
             return "CronExpression"
         else:
-            raise Exception(f"Unknown object type: {obj.__class__.__name__}")
+            raise TypeError(f"Unknown object type: {obj.__class__.__name__}")
 
 
 class ColorField(fields.String):

airflow/configuration.py

@@ -264,7 +264,7 @@ def _update_defaults_from_string(self, config_string: str):
                     )
                 self._default_values.set(section, key, value)
             if errors:
-                raise Exception(
+                raise AirflowConfigException(
                     f"The string config passed as default contains variables. "
                     f"This is not supported. String config: {config_string}"
                 )

airflow/hooks/package_index.py

@@ -73,7 +73,7 @@ def get_connection_url(self) -> Any:
         conn = self.get_connection(self.pi_conn_id)
         index_url = conn.host
         if not index_url:
-            raise Exception("Please provide an index URL.")
+            raise ValueError("Please provide an index URL.")
         return self._get_basic_auth_conn_url(index_url, conn.login, conn.password)
 
     def test_connection(self) -> tuple[bool, str]:

airflow/jobs/base_job_runner.py

@@ -19,6 +19,7 @@
 
 from typing import TYPE_CHECKING
 
+from airflow.exceptions import AirflowException
 from airflow.utils.session import NEW_SESSION, provide_session
 
 if TYPE_CHECKING:
@@ -35,7 +36,7 @@ class BaseJobRunner:
 
     def __init__(self, job: Job) -> None:
         if job.job_type and job.job_type != self.job_type:
-            raise Exception(
+            raise AirflowException(
                 f"The job is already assigned a different job_type: {job.job_type}."
                 f"This is a bug and should be reported."
             )

airflow/models/taskinstance.py

@@ -2518,7 +2518,7 @@ def _run_raw_task(
                 msg = f"Task failed due to SystemExit({e.code})"
                 self.handle_failure(msg, test_mode, context, session=session)
                 session.commit()
-                raise Exception(msg)
+                raise AirflowException(msg)
             except BaseException as e:
                 self.handle_failure(e, test_mode, context, session=session)
                 session.commit()

airflow/providers_manager.py

@@ -619,7 +619,7 @@ def _discover_all_providers_from_packages(self) -> None:
             self._provider_schema_validator.validate(provider_info)
             provider_info_package_name = provider_info["package-name"]
             if package_name != provider_info_package_name:
-                raise Exception(
+                raise ValueError(
                     f"The package '{package_name}' from setuptools and "
                     f"{provider_info_package_name} do not match. Please make sure they are aligned"
                 )

airflow/utils/dates.py

@@ -79,9 +79,9 @@ def date_range(
         return []
     if end_date:
         if start_date > end_date:
-            raise Exception("Wait. start_date needs to be before end_date")
+            raise ValueError("Wait. start_date needs to be before end_date")
         if num:
-            raise Exception("Wait. Either specify end_date OR num")
+            raise ValueError("Wait. Either specify end_date OR num")
     if not end_date and not num:
         end_date = timezone.utcnow()
 
@@ -99,7 +99,7 @@ def date_range(
     elif isinstance(delta, relativedelta):
         abs_delta = abs(delta)
     else:
-        raise Exception("Wait. delta must be either datetime.timedelta or cron expression as str")
+        raise TypeError("Wait. delta must be either datetime.timedelta or cron expression as str")
 
     dates = []
     if end_date:

airflow/utils/email.py

@@ -133,7 +133,7 @@ def send_email_smtp(
         mail_from = smtp_mail_from
     else:
         if from_email is None:
-            raise Exception(
+            raise ValueError(
                 "You should set from email - either by smtp/smtp_mail_from config or `from_email` parameter"
             )
         mail_from = from_email

airflow/utils/log/logging_mixin.py

@@ -227,7 +227,7 @@ class RedirectStdHandler(StreamHandler):
 
     def __init__(self, stream):
         if not isinstance(stream, str):
-            raise Exception(
+            raise TypeError(
                 "Cannot use file like objects. Use 'stdout' or 'stderr' as a str and without 'ext://'."
             )
 

airflow/utils/python_virtualenv.py

@@ -105,7 +105,7 @@ def prepare_virtualenv(
     execute_in_subprocess(virtualenv_cmd)
 
     if requirements is not None and requirements_file_path is not None:
-        raise Exception("Either requirements OR requirements_file_path has to be passed, but not both")
+        raise ValueError("Either requirements OR requirements_file_path has to be passed, but not both")
 
     pip_cmd = None
     if requirements is not None and len(requirements) != 0:

airflow/www/extensions/init_auth_manager.py

@@ -60,7 +60,7 @@ def init_auth_manager(appbuilder: AirflowAppBuilder) -> BaseAuthManager:
 def get_auth_manager() -> BaseAuthManager:
     """Return the auth manager, provided it's been initialized before."""
     if auth_manager is None:
-        raise Exception(
+        raise RuntimeError(
             "Auth Manager has not been initialized yet. "
             "The `init_auth_manager` method needs to be called first."
         )

tests/always/test_providers_manager.py

@@ -268,7 +268,7 @@ class MyHook:
             widgets["extra__test__my_param"] = widget_field
             widgets["my_param"] = dummy_field
         else:
-            raise Exception("unexpected")
+            raise ValueError("unexpected")
 
         provider_manager._add_widgets(
             package_name="abc",
@@ -456,9 +456,9 @@ def test_lazy_cache_dict_resolving(value, expected_outputs):
 
 def test_lazy_cache_dict_raises_error():
     def raise_method():
-        raise Exception("test")
+        raise RuntimeError("test")
 
     lazy_cache_dict = LazyDictWithCache()
     lazy_cache_dict["key"] = raise_method
-    with pytest.raises(Exception, match="test"):
+    with pytest.raises(RuntimeError, match="test"):
         _ = lazy_cache_dict["key"]

tests/api_connexion/schemas/test_common_schema.py

@@ -140,5 +140,5 @@ def test_should_serialize_cron_expression(self):
     def test_should_error_unknown_obj_type(self):
         instance = 342
         schema_instance = ScheduleIntervalSchema()
-        with pytest.raises(Exception, match="Unknown object type: int"):
+        with pytest.raises(TypeError, match="Unknown object type: int"):
             schema_instance.dump(instance)

tests/hooks/test_package_index.py

@@ -87,7 +87,7 @@ def test_get_connection_url(mock_get_connection: str | None):
         connection_url = hook_instance.get_connection_url()
         assert connection_url == expected_result
     else:
-        with pytest.raises(Exception):
+        with pytest.raises(ValueError, match="Please provide an index URL."):
             hook_instance.get_connection_url()
 
 

tests/models/test_taskinstance.py

@@ -3039,14 +3039,14 @@ def test_echo_env_variables(self, dag_maker):
     @pytest.mark.parametrize(
         "code, expected_state",
         [
-            (1, State.FAILED),
-            (-1, State.FAILED),
-            ("error", State.FAILED),
-            (0, State.SUCCESS),
-            (None, State.SUCCESS),
+            pytest.param(1, State.FAILED, id="code-positive-number"),
+            pytest.param(-1, State.FAILED, id="code-negative-number"),
+            pytest.param("error", State.FAILED, id="code-text"),
+            pytest.param(0, State.SUCCESS, id="code-zero"),
+            pytest.param(None, State.SUCCESS, id="code-none"),
         ],
     )
-    def test_handle_system_exit(self, dag_maker, code, expected_state):
+    def test_handle_system_exit_failed(self, dag_maker, code, expected_state):
         with dag_maker():
 
             def f(*args, **kwargs):
@@ -3060,10 +3060,14 @@ def f(*args, **kwargs):
         session = settings.Session()
         session.merge(ti)
         session.commit()
-        try:
+
+        if expected_state == State.SUCCESS:
+            ctx = contextlib.nullcontext()
+        else:
+            ctx = pytest.raises(AirflowException, match=rf"Task failed due to SystemExit\({code}\)")
+
+        with ctx:
             ti._run_raw_task()
-        except Exception:
-            ...
         ti.refresh_from_db()
         assert ti.state == expected_state
 

tests/operators/test_python.py

@@ -748,14 +748,14 @@ def test_template_fields(self):
 
     def test_fail(self):
         def f():
-            raise Exception
+            raise RuntimeError
 
         with pytest.raises(CalledProcessError):
             self.run_as_task(f)
 
     def test_fail_with_message(self):
         def f():
-            raise Exception("Custom error message")
+            raise RuntimeError("Custom error message")
 
         with pytest.raises(AirflowException, match="Custom error message"):
             self.run_as_task(f)
@@ -765,7 +765,7 @@ def f():
             global virtualenv_string_args
             print(virtualenv_string_args)
             if virtualenv_string_args[0] != virtualenv_string_args[2]:
-                raise Exception
+                raise RuntimeError
 
         self.run_as_task(f, string_args=[1, 2, 1])
 
@@ -774,7 +774,7 @@ def f(a, b, c=False, d=False):
             if a == 0 and b == 1 and c and not d:
                 return True
             else:
-                raise Exception
+                raise RuntimeError
 
         self.run_as_task(f, op_args=[0, 1], op_kwargs={"c": True})
 
@@ -940,7 +940,7 @@ def f():
                 import funcsigs  # noqa: F401
             except ImportError:
                 return True
-            raise Exception
+            raise RuntimeError
 
         self.run_as_task(f, system_site_packages=False, requirements=["dill"])
 
@@ -955,7 +955,7 @@ def f():
             import funcsigs
 
             if funcsigs.__version__ != "0.4":
-                raise Exception
+                raise RuntimeError
 
         self.run_as_task(f, requirements=["funcsigs==0.4"])
 
@@ -971,7 +971,7 @@ def f():
             import funcsigs
 
             if funcsigs.__version__ != "0.4":
-                raise Exception
+                raise RuntimeError
 
         self.run_as_task(f, requirements=["funcsigs==0.4"], do_not_use_caching=True)
 
@@ -1038,7 +1038,7 @@ def f():
                 {}.iteritems()
             except AttributeError:
                 return
-            raise Exception
+            raise RuntimeError
 
         self.run_as_task(f, python_version="3", use_dill=False, requirements=["dill"])
 
@@ -1274,7 +1274,7 @@ def f(a, b, c=False, d=False):
             if a == 0 and b == 1 and c and not d:
                 return True
             else:
-                raise Exception
+                raise RuntimeError
 
         with pytest.raises(AirflowException, match="but got 'bool'"):
             self.run_as_task(f, op_args=[0, 1], op_kwargs={"c": True})

tests/utils/test_dates.py

@@ -109,16 +109,16 @@ def test_no_delta(self):
         assert dates.date_range(datetime(2016, 1, 1), datetime(2016, 1, 3)) == []
 
     def test_end_date_before_start_date(self):
-        with pytest.raises(Exception, match="Wait. start_date needs to be before end_date"):
+        with pytest.raises(ValueError, match="Wait. start_date needs to be before end_date"):
             dates.date_range(datetime(2016, 2, 1), datetime(2016, 1, 1), delta=timedelta(seconds=1))
 
     def test_both_end_date_and_num_given(self):
-        with pytest.raises(Exception, match="Wait. Either specify end_date OR num"):
+        with pytest.raises(ValueError, match="Wait. Either specify end_date OR num"):
             dates.date_range(datetime(2016, 1, 1), datetime(2016, 1, 3), num=2, delta=timedelta(seconds=1))
 
     def test_invalid_delta(self):
         exception_msg = "Wait. delta must be either datetime.timedelta or cron expression as str"
-        with pytest.raises(Exception, match=exception_msg):
+        with pytest.raises(TypeError, match=exception_msg):
             dates.date_range(datetime(2016, 1, 1), datetime(2016, 1, 3), delta=1)
 
     def test_positive_num_given(self):