**Apache Airflow version**:
2.0 / master

**Environment**:

breeze

**What happened**:

Using this DAG:

```py
from airflow.operators.bash import BashOperator
from airflow.operators.python import task
from airflow.models import DAG
from airflow.utils.task_group import TaskGroup

@task
def show():
    print("Cats are awesome!")

with DAG(
    "using_task_group",
    default_args={'owner': 'airflow'},
    start_date=days_ago(2),
    schedule_interval=None,
) as dag3:
    start_task = BashOperator(
        task_id="start_task",
        bash_command="echo start",
    )

    end_task = BashOperator(
        task_id="end_task",
        bash_command="echo end",
    )

    with TaskGroup(group_id="show_tasks") as tg1:
        previous_show = show()
        for _ in range(100):
            next_show = show()
            previous_show >> next_show
            previous_show = next_show
```
I get:
```
Broken DAG: [/files/dags/test.py] Traceback (most recent call last):
  File "/opt/airflow/airflow/models/baseoperator.py", line 410, in __init__
    task_group.add(self)
  File "/opt/airflow/airflow/utils/task_group.py", line 140, in add
    raise DuplicateTaskIdFound(f"Task id '{key}' has already been added to the DAG")
airflow.exceptions.DuplicateTaskIdFound: Task id 'show_tasks.show' has already been added to the DAG
```

If I remove the task group the task are generated as expected.

**What you expected to happen**:

I expect to be able to generate tasks dynamically using TaskGroup and task decoratos.

**How to reproduce it**:

Use the DAG from above.

**Anything else we need to know**:

N/A

tests/operators/test_python.py
#
#
#
import copy
import logging
import sys
import unittest.mock
from collections import namedtuple
from datetime import date, datetime, timedelta
from subprocess import CalledProcessError
from typing import List
import funcsigs
import pytest
from airflow.exceptions import AirflowException