You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/09/23 14:28:00 UTC
[GitHub] [airflow] eladkal opened a new pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
eladkal opened a new pull request #18471:
URL: https://github.com/apache/airflow/pull/18471
closes: https://github.com/apache/airflow/issues/18454
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
josh-fell commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925948392
Perhaps but the ShortCircuitOperator can be redundant here if we're skipping downstream tasks with this operator now. Seconded on @uranusjr's suggestion as well FWIW.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-926071182
If you want to skip everything you can `raise AirflowSkipException` I think. (that would have a slightly different behaviour of also skipping the branch op itself, unless we caught that in this op and turned it in to skip all downstream)
That's all a separate pr/feature though
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925914848
> I could understand users may not want the DAG to fail outright if the branching section is not critical path like with some feature engineering use cases I've seen. It may be more intuitive to not have to create an empty branch to simply continue with the pipeline if an output doesn't need to be tracked. An edge case for sure with an obvious workaround.
>
> Could this be a configurable mode: fail or skip if no branches found?
This sounds more like a wrong usage of branching. For conditional junctions we have ShortCircuitOperator
but i'm OK with @uranusjr suggestion if this is prefered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r715913452
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,16 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if not isinstance(branch, (str, List)):
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ branches = {branch}
Review comment:
```suggestion
if isinstance(branch, str):
branches = {branch}
elif isinstance(branch, list):
branches = set(branch)
else:
raise AirflowException("Branch callable must return either a task ID or a list of IDs")
```
Otherwise if `branch` is a list (e.g. `["task_a", "task_b"]`), `branches` would become a set containing one list of strings (`{["task_a", "task_b"]}`). We need a set of strings (`{"task_a", "task_b"}`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r718830238
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
Ah yeah. Ill raisr a PR to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r716030104
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
Review comment:
Should we also change SkipMixin while we are changing it ? (per the comment above)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925878936
Instead of explicitly forbidding this, would it be reasonable to instead handle the None case and skip all the downstreams instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r718830238
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
Ah yeah. Ill raisr a PR to fix it.
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
Ah yeah. Ill raise a PR to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-926487236
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-926425630
> If you want to skip everything you can `raise AirflowSkipException` I think. (that would have a slightly different behaviour of also skipping the branch op itself, unless we caught that in this op and turned it in to skip all downstream)
So this is a possible route. We will need to keep the change in `SkipMixin` itself to avoid changing all the branch operators.
> That's all a separate pr/feature though
I agree. This PR is just replacing the error message to a more meaningful one. There is no change in functionality here.
I'll work on trying to solve the issue from the root for a future PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal edited a comment on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal edited a comment on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925980093
@uranusjr correct me if I'm wrong but this requires rewriting [skip_all_except](https://github.com/apache/airflow/blob/958b679dae6bbd9def7c60191c3a7722ce81382a/airflow/models/skipmixin.py#L134)
The way I see it I can't just blindly skip all downstream tasks as unlike `ShortCircuitOperator` branching does respect trigger rules so I need to skip all downstream tasks until a join/junction task is found.
Since my input is None we need to search if such junction exists.
So I need to set difference between two cases:
If I have a case of:
# branch -----> task2
# \
# \
# task1
Then I need to execute [skip](https://github.com/apache/airflow/blob/958b679dae6bbd9def7c60191c3a7722ce81382a/airflow/models/skipmixin.py#L66)
but If I have a case like:
# branch -----> task2 <----- task3
# \
# \
# task1
Then I need to run `skip_all_except(context['ti'],['task2])` because task2 may still run depends on its `trigger_rule`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r716037615
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
Review comment:
Seperate PR.
I will work on it. We want also to change the functionality so it will accept None and skip all downstream tasks (untill a junction/join task).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925980093
@uranusjr correct me if I'm wrong but this requires rewriting [skip_all_except](https://github.com/apache/airflow/blob/958b679dae6bbd9def7c60191c3a7722ce81382a/airflow/models/skipmixin.py#L134)
The way I see it I can't just blindly skip all downstream tasks as unlike `ShortCircuitOperator` branching does respect trigger rules so I need to skip all downstream tasks until a join/junction task is found.
Since my input is None it's up to the code to search if such junction exists.
So I need to set difference between two cases:
If I have a case of:
# branch -----> task2
# \
# \
# task1
Then I need to execute [skip](https://github.com/apache/airflow/blob/958b679dae6bbd9def7c60191c3a7722ce81382a/airflow/models/skipmixin.py#L66)
but If I have a case like:
# branch -----> task2 <----- task3
# \
# \
# task1
Then I need to run `skip_all_except(context['ti'],['task2])` because task2 may still run depends on its `trigger_rule`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal merged pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal merged pull request #18471:
URL: https://github.com/apache/airflow/pull/18471
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell edited a comment on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
josh-fell edited a comment on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925902196
> > Instead of explicitly forbidding this, would it be reasonable to handle the None case and skip all the downstreams instead?
>
> When would None be desired?
I could understand users may not want the DAG to fail outright if the branching section is not critical path like with some feature engineering use cases I've seen. It may be more intuitive to not have to create an empty branch to simply continue with the pipeline if an output doesn't need to be tracked. An edge case for sure with an obvious workaround.
Could this be a configurable mode: fail or skip if no branches found?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r719100401
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
josh-fell commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925902196
> > Instead of explicitly forbidding this, would it be reasonable to handle the None case and skip all the downstreams instead?
>
> When would None be desired?
I could understand users may not want the DAG to fail outright if the branching section is not critical path like with some feature engineering use cases I've seen. It may be more intuitive to not have to create an empty branch to simply continue with the pipeline if an output doesn't need to be tracked.
Could this be a configurable mode: fail or skip if no branches found?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r719047738
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
https://github.com/apache/airflow/pull/18623
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr edited a comment on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
uranusjr edited a comment on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925878936
Instead of explicitly forbidding this, would it be reasonable to handle the None case and skip all the downstreams instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] MatrixManAtYrService commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
MatrixManAtYrService commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925948648
I had the same thought as @uranusjr, skipping all downstream tasks seems like an obvious way to interpret `None` to me, like an `if` statement with no `else`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r718830238
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
Ah yeah. Ill raise a PR to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r716030104
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
Review comment:
Should we also change SkipMixin while we are changing it (or will it be a separate PR) ? (per the comment above)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r715463290
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,8 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ if branch is None:
+ raise AirflowException("Branch callable must return at least one task_id to follow")
Review comment:
Kind of wondering whether we should check for all non-str values. Or even check the value is a valid task ID? Something like…
```python
if isinstance(branch, str):
branches = [branch]
try:
branches = list(branch) # Check iterability.
except TypeError:
raise AirflowException("Branch callable must return either a task ID or a list of IDs")
valid_task_ids = set(context["context"].task_ids)
if not any(b in valid_task_ids for b in branches):
raise AirflowException("Branch callable must return at least one task ID to follow")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] MatrixManAtYrService edited a comment on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
MatrixManAtYrService edited a comment on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925948648
I had the same thought as @uranusjr, skipping all downstream tasks seems like an obvious way to interpret `None` to me, like an `if` statement with no `else`.
I'm not gonna fight for it though. Scope creep is real and I'll be happy enough to have a better error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r718826421
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
@eladkal Did you mean something like
`Branch callable must *return* valid task_ids`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r715463290
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,8 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ if branch is None:
+ raise AirflowException("Branch callable must return at least one task_id to follow")
Review comment:
Kind of wondering whether we should check for all non-str values. Or even check the value is a valid task ID? Something like…
```python
if isinstance(branch, str):
branches = [branch]
try:
branches = list(branch) # Check iterability.
except TypeError:
raise AirflowException("Branch callable must return either a task ID or a list of IDs")
valid_task_ids = set(context["dag"].task_ids)
if not any(b in valid_task_ids for b in branches):
raise AirflowException("Branch callable must return at least one task ID to follow")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r716030104
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
Review comment:
Should we also change SkipMixin while we are changing it ? (see the comment above)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r715621156
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,8 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ if branch is None:
+ raise AirflowException("Branch callable must return at least one task_id to follow")
Review comment:
yeah this is something I had in mind - added it in last commit
In general this is something we should have in skipmixin as it's logic shared by all branch operators.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
eladkal commented on pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#issuecomment-925880055
> Instead of explicitly forbidding this, would it be reasonable to handle the None case and skip all the downstreams instead?
When would None be desired?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] edwardwang888 commented on a change in pull request #18471: Improve error message for BranchPythonOperator when no task_id to follow
Posted by GitBox <gi...@apache.org>.
edwardwang888 commented on a change in pull request #18471:
URL: https://github.com/apache/airflow/pull/18471#discussion_r718826421
##########
File path: airflow/operators/python.py
##########
@@ -178,6 +178,19 @@ class BranchPythonOperator(PythonOperator, SkipMixin):
def execute(self, context: Dict):
branch = super().execute(context)
+ # TODO: The logic should be moved to SkipMixin to be available to all branch operators.
+ if isinstance(branch, str):
+ branches = {branch}
+ elif isinstance(branch, list):
+ branches = set(branch)
+ else:
+ raise AirflowException("Branch callable must return either a task ID or a list of IDs")
+ valid_task_ids = set(context["dag"].task_ids)
+ invalid_task_ids = branches - valid_task_ids
+ if invalid_task_ids:
+ raise AirflowException(
+ f"Branch callable must valid task_ids. Invalid tasks found: {invalid_task_ids}"
Review comment:
@eladkal Did you mean something like
`Branch callable must *return* valid task_ids`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org