BREAKING_CHANGE: remove airflow.operators.email.EmailOperator in favo…

…r of SMTP provider operator (#46573)

* BREAKING_CHANGE: remove airflow.operators.email.EmailOperator in favor of SMTP provider operator

* Update newsfragments/46572.significant.rst

Co-authored-by: Wei Lee <[email protected]>

* fix static checks

---------

Co-authored-by: Wei Lee <[email protected]>

airflow/example_dags/example_dag_decorator.py

@@ -24,7 +24,7 @@
 
 from airflow.decorators import dag, task
 from airflow.models.baseoperator import BaseOperator
-from airflow.operators.email import EmailOperator
+from airflow.providers.standard.operators.bash import BashOperator
 
 if TYPE_CHECKING:
     from airflow.sdk.definitions.context import Context
@@ -48,27 +48,24 @@ def execute(self, context: Context):
     catchup=False,
     tags=["example"],
 )
-def example_dag_decorator(email: str = "[email protected]"):
+def example_dag_decorator(url: str = "http://httpbin.org/get"):
     """
-    DAG to send server IP to email.
+    DAG to get IP address and echo it via BashOperator.
 
-    :param email: Email to send IP to. Defaults to [email protected].
+    :param url: URL to get IP address from. Defaults to "http://httpbin.org/get".
     """
-    get_ip = GetRequestOperator(task_id="get_ip", url="http://httpbin.org/get")
+    get_ip = GetRequestOperator(task_id="get_ip", url=url)
 
     @task(multiple_outputs=True)
-    def prepare_email(raw_json: dict[str, Any]) -> dict[str, str]:
+    def prepare_command(raw_json: dict[str, Any]) -> dict[str, str]:
         external_ip = raw_json["origin"]
         return {
-            "subject": f"Server connected from {external_ip}",
-            "body": f"Seems like today your server executing Airflow is connected from IP {external_ip}<br>",
+            "command": f"echo 'Seems like today your server executing Airflow is connected from IP {external_ip}'",
         }
 
-    email_info = prepare_email(get_ip.output)
+    command_info = prepare_command(get_ip.output)
 
-    EmailOperator(
-        task_id="send_email", to=email, subject=email_info["subject"], html_content=email_info["body"]
-    )
+    BashOperator(task_id="echo_ip_info", bash_command=command_info["command"])
 
 
 example_dag = example_dag_decorator()

docs/apache-airflow/core-concepts/operators.rst

@@ -30,7 +30,6 @@ Airflow has a very extensive set of operators available, with some built-in to t
 
 - :class:`~airflow.providers.standard.operators.bash.BashOperator` - executes a bash command
 - :class:`~airflow.providers.standard.operators.python.PythonOperator` - calls an arbitrary Python function
-- :class:`~airflow.operators.email.EmailOperator` - sends an email
 - Use the ``@task`` decorator to execute an arbitrary Python function. It doesn't support rendering jinja templates passed as arguments.
 
 .. note::
@@ -41,6 +40,7 @@ For a list of all core operators, see: :doc:`Core Operators and Hooks Reference
 
 If the operator you need isn't installed with Airflow by default, you can probably find it as part of our huge set of community :doc:`provider packages <apache-airflow-providers:index>`. Some popular operators from here include:
 
+- :class:`~airflow.providers.smtp.operators.smtp.EmailOperator`
 - :class:`~airflow.providers.http.operators.http.HttpOperator`
 - :class:`~airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator`
 - :class:`~airflow.providers.docker.operators.docker.DockerOperator`

docs/apache-airflow/core-concepts/taskflow.rst

@@ -25,7 +25,7 @@ If you write most of your DAGs using plain Python code rather than Operators, th
 TaskFlow takes care of moving inputs and outputs between your Tasks using XComs for you, as well as automatically calculating dependencies - when you call a TaskFlow function in your DAG file, rather than executing it, you will get an object representing the XCom for the result (an ``XComArg``), that you can then use as inputs to downstream tasks or operators. For example::
 
     from airflow.decorators import task
-    from airflow.operators.email import EmailOperator
+    from airflow.providers.email import EmailOperator
 
     @task
     def get_ip():

docs/apache-airflow/operators-and-hooks-ref.rst

@@ -56,9 +56,6 @@ For details see: :doc:`apache-airflow-providers:operators-and-hooks-ref/index`.
    * - :mod:`airflow.providers.standard.operators.empty`
      -
 
-   * - :mod:`airflow.operators.email`
-     -
-
    * - :mod:`airflow.operators.generic_transfer`
      -
 

newsfragments/46572.significant.rst

@@ -0,0 +1,22 @@
+In Airflow 3.0, ``airflow.operators.email.EmailOperator`` is removed.
+
+Instead, users can install ``smtp`` provider and import ``EmailOperator`` from the the module ``airflow.providers.smtp.operators.smtp``.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [ ] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [ ] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [x] Code interface changes
+
+* Migration rules needed
+
+  * ruff
+
+    * <RULE_ID>
+
+      * [ ] ``airflow.operators.email`` → ``airflow.providers.smtp.operators.smtp.EmailOperator``