jason810496 commented on code in PR #74057:
URL: https://github.com/apache/airflow/pull/74057#discussion_r4163050220
##########
go-sdk/airflow/dag.go:
##########
@@ -173,21 +173,20 @@ func (d *DagRef) Task(fn any, opts ...TaskOption)
*TaskRef {
d.dagID, i, opt,
))
}
- opt.applyTask(&cfg)
- }
- if len(cfg.specs) > 1 {
- panic(fmt.Sprintf(
- "airflow.DagRef.Task: task %q of Dag %q got %d
airflow.TaskSpec values; "+
- "set all of the task's attributes in one
TaskSpec",
- findTaskName(fn, cfg.specs), d.dagID, len(cfg.specs),
- ))
+ if err := opt.applyTask(&cfg); err != nil {
+ // A task from TriggerDagRun takes no Inputs at all,
and the check after this loop
+ // reports that. err says to merge the Inputs into one,
which would not fix a task from
+ // TriggerDagRun.
+ if _, ok := opt.(inputs); ok && isTrigger {
+ continue
+ }
+ panic(fmt.Sprintf(
+ "airflow.DagRef.Task: task %q of Dag %q: %v",
findTaskName(fn, opts), d.dagID, err,
+ ))
+ }
}
- var spec TaskSpec
- if len(cfg.specs) == 1 {
- spec = cfg.specs[0]
- }
- taskID := spec.TaskID
+ taskID := cfg.spec.TaskID
Review Comment:
Heads-up: #74064 adds `checkTaskSpec(spec)` further down in this function.
With this change removing the local `spec`, the two PRs merge with no conflict
but `go-sdk/airflow` no longer compiles. Whichever lands second needs
`checkTaskSpec(cfg.spec)`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
