Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix: Added example of _push_xcoms_if_necessary to deferring.rst #45158

Open
wants to merge 6 commits into
base: main
Choose a base branch
from

Conversation

avyuktsoni0731
Copy link

@avyuktsoni0731 avyuktsoni0731 commented Dec 22, 2024


What is the purpose of this PR?

This PR updates the deferring.rst documentation to include an example of the _push_xcoms_if_necessary method, as suggested in issue #44759.

The new content provides guidance on:

  • Pushing XComs when deferred tasks exit directly from triggers.
  • How to integrate _push_xcoms_if_necessary with custom triggers.

Related Issues

Closes #44759.

Changes Made

  • Added a new section under Exiting deferred task from Triggers in deferring.rst to describe the _push_xcoms_if_necessary method and its usage.
  • Included example code snippets to illustrate the implementation.

Checklist

  • Documentation updated.
  • Code quality checks passed (pre-commit hooks run).

Reviewers

Please suggest any further refinements or additional examples if required.

^ 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.

Copy link

boring-cyborg bot commented Dec 22, 2024

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

@potiuk
Copy link
Member

potiuk commented Dec 22, 2024

@tirkarthi @eladkal -> I think you were more involved in #44759 - that one looks nice and good I will have a few nits.

async def run(self) -> AsyncIterator[TriggerEvent]:
await asyncio.sleep(self.duration.total_seconds())
if self.end_from_trigger:
task_instance = ... # Get the relevant TaskInstance
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm. I am not sure how to get the task instance here . I am not sure if you can do it automatically ?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, let me look into this again, and try to make the necessary changes that @tirkarthi suggested.

@tirkarthi
Copy link
Contributor

I don't think this is the right approach. My comment was that you can set self.xcoms which is a dictionary and then the triggerer pushes it to xcom once the event is emitted. There is no need to reimplement _push_xcoms_if_necessary or even mention it as a public interface. The example would be to set self.xcoms and explain to users that triggerer does the job of pushing them to xcom.

#44759 (comment)

@avyuktsoni0731
Copy link
Author

@potiuk @tirkarthi I've created a PR with the changes, please do look into it, and let me know if I'm still not able to understand it properly, will make the required changes if needed.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

Successfully merging this pull request may close these issues.

Trigger event from deferred task does not get scheduled immediately, leading to timeout.
3 participants