def notify(self, context):

     with SmtpHook(smtp_conn_id=self.smtp_conn_id) as smtp_hook:

         """Send a email via smtp server."""
         fields_to_re_render = []
         if self.from_email is None:
             if smtp_hook.from_email is not None:
                 self.from_email = smtp_hook.from_email
             else:
                 raise ValueError("You should provide `from_email` or define it in the connection")
             fields_to_re_render.append("from_email")
         if self.subject is None:
             smtp_default_templated_subject_path: str
             if smtp_hook.subject_template:
                 smtp_default_templated_subject_path = smtp_hook.subject_template
             else:
                 smtp_default_templated_subject_path = (
                     Path(__file__).parent / "templates" / "email_subject.jinja2"
                 ).as_posix()
             self.subject = self._read_template(smtp_default_templated_subject_path)
             fields_to_re_render.append("subject")
         if self.html_content is None:
             smtp_default_templated_html_content_path: str
             if smtp_hook.html_content_template:
                 smtp_default_templated_html_content_path = smtp_hook.html_content_template
             else:
                 smtp_default_templated_html_content_path = (
                     Path(__file__).parent / "templates" / "email.html"
                 ).as_posix()
             self.html_content = self._read_template(smtp_default_templated_html_content_path)
             fields_to_re_render.append("html_content")
         if fields_to_re_render:
             jinja_env = self.get_template_env(dag=context["dag"])
             self._do_render_template_fields(self, fields_to_re_render, context, jinja_env, set())

         smtp_hook.send_email_smtp(
             smtp_conn_id=self.smtp_conn_id,
             from_email=self.from_email,
             to=self.to,
             subject=self.subject,
             html_content=self.html_content,
             files=self.files,
             cc=self.cc,
             bcc=self.bcc,
             mime_subtype=self.mime_subtype,
             mime_charset=self.mime_charset,
             custom_headers=self.custom_headers,
         )

UndefinedError: 'airflow.sdk.execution_time.task_runner.RuntimeTaskInstance object' has no attribute 'state'
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/sdk/bases/notifier.py", line 105 in __call__
File "/opt/airflow/venv/lib/python3.11/site-packages/smtp_thehub_provider/notifications/smtp.py", line 142 in notify
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/sdk/definitions/_internal/templater.py", line 121 in _do_render_template_fields
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/sdk/definitions/_internal/templater.py", line 177 in render_template
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/sdk/bases/notifier.py", line 57 in _render
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/sdk/definitions/_internal/templater.py", line 133 in _render
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/utils/helpers.py", line 244 in render_template_to_string
File "/opt/airflow/venv/lib/python3.11/site-packages/airflow/utils/helpers.py", line 239 in render_template
File "<template>", line 29 in root
File "/opt/airflow/venv/lib/python3.11/site-packages/jinja2/runtime.py", line 859 in _fail_with_undefined_error

from airflow import DAG
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from datetime import datetime

default_args = {
    "owner": "airflow",
}

with DAG(
    dag_id="airflow_email_test",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    default_args=default_args,
    tags=["maintenance", "cleanup"],
) as dag:

    clean_logs = EmptyOperator(
        task_id="empty_operator",
        on_success_callback=SmtpNotifier(from_email='[email protected], to="[email protected]"),
    )