Add listeners for Dag import errors

[pushpendu91]

---

This PR expands listeners API to be notified about DAG Import error when updating the import_errors table.

At present to identify DAG import errors, user have to refresh the UI or use Importerror REST API to get the import error details. And as there is no definite timeline( by next dag_parsing_interval) when the DAG will be parsed and import errors will be raised users need to keep on refreshing the UI or hit the API.

With this PR Airflow itself will notify the listener about the addition of import error or existing import errors. And using the below events user can take action accordingly.

a. on_new_dag_import_error
b. on_existing_dag_import_error

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

[amoghrajesh]

I would like to see some unit tests for this change as it is a change in behaviour.

Also, some expert in scheduler area to review it @uranusjr at the very least

[amoghrajesh]

Approved the workflow run also, watch out for CI results

[pushpendu91]

@amoghrajesh Thank you for reviewing the PR. I was trying to add test cases, but was not able to find an way how to pass a broken DAG . If someone can guide me on the same, that will be helpful.

[pushpendu91]

@amoghrajesh @uranusjr I have added the unit tests for the change and below is the test result from my local machine.
I kindly request you to review the change.

root@e316901b91ee:/opt/airflow# pytest tests/listeners/test_dag_import_error_listener.py
====================================================== test session starts ======================================================
platform linux -- Python 3.8.19, pytest-8.2.1, pluggy-1.5.0 -- /usr/local/bin/python
cachedir: .pytest_cache
rootdir: /opt/airflow
configfile: pyproject.toml
plugins: cov-5.0.0, rerunfailures-14.0, mock-3.14.0, custom-exit-code-0.3.0, timeouts-1.2.1, requests-mock-1.12.1, icdiff-0.9, time-machine-2.14.1, xdist-3.6.1, instafail-0.5.0, anyio-4.3.0, asyncio-0.23.7
asyncio: mode=strict
setup timeout: 0.0s, execution timeout: 0.0s, teardown timeout: 0.0s
collected 2 items                                                                                                               

tests/listeners/test_dag_import_error_listener.py::TestDagFileProcessor::test_newly_added_import_error PASSED             [ 50%]
tests/listeners/test_dag_import_error_listener.py::TestDagFileProcessor::test_already_existing_import_error PASSED        [100%]
============================================= Warning summary. Total: 1, Unique: 1 ==============================================
tests: total 1, unique 1
  runtest: total 1, unique 1
Warnings saved into /opt/airflow/tests/warnings.txt file.

======================================================= 2 passed in 2.79s =======================================================
root@e316901b91ee:/opt/airflow# 

[uranusjr]

Implementation looks OK to me at first glance.

[amoghrajesh]

Looks good to me. @uranusjr WDYT?

[eladkal]

@pushpendu91 can you fix static checks?

[pushpendu91]

@eladkal I have fixed the static checks errors. Kindly request you to review the change.

[pushpendu91]

@amoghrajesh @eladkal Request you to please start the workflow, so that static checks can be validated. I ran in my local using pre-commit but I was not getting any error.

[pushpendu91]

@amoghrajesh request you to please check the workflow. As it is not giving error in my local I am relying on the actual workflow here in github.

[eladkal]

Can you also add the needed doc entry in https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/listeners.html

[pushpendu91]

@eladkal I have added the required documentation. Thank you.

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.