Use task_id as backup if no task.label

[bbovenzi]

Fixes: #33093

---

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

[uranusjr]

Where does label come from? Maybe we should fix that at the Python side instead.

[bbovenzi]

@uranusjr I agree. I just don't know why label isn't being populated. @dstandish With your example dag, do you have any ideas?

[dstandish]

I was about to say that I think we can close this because after #33102, setting deps in a context no longer "auto adds" tasks to the context. So now the sample dag posted in #33093 will produce this graph:

I.e. nothing will be in the task group.

BUT...

then i added the tasks to the context with add_task:

    with TaskGroup("my_group") as tg:
        setup1 >> work1 >> teardown1
        setup1 >> teardown1
        tg.add_task(setup1)
        tg.add_task(teardown1)
        tg.add_task(work1)
    tg >> work2

And now we encounter the issue again:

I'll poke around and see if I can figure anything out.

[dstandish]

Ok so the problem is this logic:

    @property
    def label(self) -> str | None:
        tg = self.task_group
        if tg and tg.node_id and tg.prefix_group_id:
            # "task_group_id.task_id" -> "task_id"
            return self.node_id[len(tg.node_id) + 1 :]
        return self.node_id

The issue is when the task is instantiated outside of the group, it doesn't get prefixed. So self.node_id[len(tg.node_id) + 1 :] is ''

[dstandish]

An odd consequence of this is if the task group id is short, we'll actually get a partial name:

code to repro:

with DAG(dag_id="leaves_ignored", start_date=pendulum.now()):

    @task
    def setup1():
        ...

    @task
    def setup2():
        ...

    @task
    def teardown1():
        ...

    @task
    def teardown2():
        ...

    @task
    def work1():
        ...

    @task
    def work2():
        ...


    setup1 = setup1()
    # setup2 = setup2()
    teardown1 = teardown1()
    # teardown2 = teardown2()
    work1 = work1()
    work2 = work2()
    with TaskGroup("tg") as tg:
        setup1 >> work1 >> teardown1
        setup1 >> teardown1
        tg.add_task(setup1)
        tg.add_task(teardown1)
        tg.add_task(work1)
    tg >> work2

Maybe we should remove add_task for task group for now, wdyt @ephraimbuddy @uranusjr? It still has not been released yet.

[uranusjr]

Removing add_task sounds reasonable, I think that’s proposed once previously for another reason as well?

[ephraimbuddy]

Removing add_task sounds reasonable, I think that’s proposed once previously for another reason as well?

Yeah. We should remove it. We previously wanted to make it add_tasks