We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Error setting dependencies on task_group defined using the decorator
### Apache Airflow version

2.2.2 (latest released)

### Operating System

MacOS 11.6.1

### Versions of Apache Airflow Providers

$ pip freeze | grep airflow
apache-airflow==2.2.2
apache-airflow-providers-celery==2.1.0
apache-airflow-providers-ftp==2.0.1
apache-airflow-providers-http==2.0.1
apache-airflow-providers-imap==2.0.1
apache-airflow-providers-sqlite==2.0.1

### Deployment

Other

### Deployment details

`airflow standalone`

### What happened

```
AttributeError: 'NoneType' object has no attribute 'update_relative'
```

### What you expected to happen

Task group should be set as downstream of `start` task, and upstream of `end` task

### How to reproduce

* Add the following code to dags folder

```python
from datetime import datetime

from airflow.decorators import dag, task, task_group


@dag(start_date=datetime(2023, 1, 1), schedule_interval="@once")
def test_dag_1():
    @task
    def start():
        pass

    @task
    def do_thing(x):
        print(x)

    @task_group
    def do_all_things():
        do_thing(1)
        do_thing(2)

    @task
    def end():
        pass

    start() >> do_all_things() >> end()


test_dag_1()
```

* Run `airflow standalone`

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/decorators/task_group.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """
19 A TaskGroup is a collection of closely related tasks on the same DAG that should be grouped
20 together when the DAG is displayed graphically.
21 """
22 import functools
23 import warnings
24 from inspect import signature
25 from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Optional, TypeVar, cast, overload
26 
27 import attr
28 
29 from airflow.utils.task_group import MappedTaskGroup, TaskGroup
30 
31 if TYPE_CHECKING:
32     from airflow.models import DAG
33 
34 F = TypeVar("F", bound=Callable[..., Any])
35 T = TypeVar("T", bound=Callable)
36 R = TypeVar("R")
37 
38 task_group_sig = signature(TaskGroup.__init__)
39 
40 
41 @attr.define
42 class TaskGroupDecorator(Generic[R]):
43     """:meta private:"""
44 
45     function: Callable[..., R] = attr.ib(validator=attr.validators.is_callable())
46     kwargs: Dict[str, Any] = attr.ib(factory=dict)
47     """kwargs for the TaskGroup"""
48 
49     @function.validator
50     def _validate_function(self, _, f):
51         if 'self' in signature(f).parameters:
52             raise TypeError('@task_group does not support methods')
53 
54     @kwargs.validator
55     def _validate(self, _, kwargs):
56         task_group_sig.bind_partial(**kwargs)
57 
58     def __attrs_post_init__(self):
59         self.kwargs.setdefault('group_id', self.function.__name__)
60 
61     def _make_task_group(self, **kwargs) -> TaskGroup:
62         return TaskGroup(**kwargs)
63 
64     def __call__(self, *args, **kwargs) -> R:
65         with self._make_task_group(add_suffix_on_collision=True, **self.kwargs):
66             # Invoke function to run Tasks inside the TaskGroup
67             return self.function(*args, **kwargs)
68 
69     def partial(self, **kwargs) -> "MappedTaskGroupDecorator[R]":
70         return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).partial(**kwargs)
71 
72     def map(self, **kwargs) -> R:
73         return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).map(**kwargs)
74 
75 
76 @attr.define
77 class MappedTaskGroupDecorator(TaskGroupDecorator[R]):
78     """:meta private:"""
79 
80     partial_kwargs: Dict[str, Any] = attr.ib(factory=dict)
81     """static kwargs for the decorated function"""
82     mapped_kwargs: Dict[str, Any] = attr.ib(factory=dict)
83     """kwargs for the decorated function"""
84 
85     def __call__(self, *args, **kwargs):
86         raise RuntimeError("A mapped @task_group cannot be called. Use `.map` and `.partial` instead")
87 
88     def _make_task_group(self, **kwargs) -> MappedTaskGroup:
89         tg = MappedTaskGroup(**kwargs)
90         tg.partial_kwargs = self.partial_kwargs
91         tg.mapped_kwargs = self.mapped_kwargs
92         return tg
93 
94     def partial(self, **kwargs) -> "MappedTaskGroupDecorator[R]":
95         if self.partial_kwargs:
96             raise RuntimeError("Already a partial task group")
97         self.partial_kwargs.update(kwargs)
98         return self
99 
100     def map(self, **kwargs) -> R:
101         if self.mapped_kwargs:
102             raise RuntimeError("Already a mapped task group")
103         self.mapped_kwargs = kwargs
104 
105         call_kwargs = self.partial_kwargs.copy()
106         duplicated_keys = set(call_kwargs).intersection(kwargs)
107         if duplicated_keys:
108             raise RuntimeError(f"Cannot map partial arguments: {', '.join(sorted(duplicated_keys))}")
109         call_kwargs.update({k: object() for k in kwargs})
110 
111         return super().__call__(**call_kwargs)
112 
113     def __del__(self):
114         if not self.mapped_kwargs:
115             warnings.warn(f"Partial task group {self.function.__name__} was never mapped!")
116 
117 
118 # This covers the @task_group() case. Annotations are copied from the TaskGroup
119 # class, only providing a default to 'group_id' (this is optional for the
120 # decorator and defaults to the decorated function's name). Please keep them in
121 # sync with TaskGroup when you can! Note that since this is an overload, these
122 # argument defaults aren't actually used at runtime--the real implementation
123 # does not use them, and simply rely on TaskGroup's defaults, so it's not
124 # disastrous if they go out of sync with TaskGroup.
125 @overload
126 def task_group(
127     group_id: Optional[str] = None,
128     prefix_group_id: bool = True,
129     parent_group: Optional["TaskGroup"] = None,
130     dag: Optional["DAG"] = None,
131     default_args: Optional[Dict[str, Any]] = None,
132     tooltip: str = "",
133     ui_color: str = "CornflowerBlue",
134     ui_fgcolor: str = "#000",
135     add_suffix_on_collision: bool = False,
136 ) -> Callable[[F], F]:
137     ...
138 
139 
140 # This covers the @task_group case (no parentheses).
141 @overload
142 def task_group(python_callable: F) -> F:
143     ...
144 
145 
146 def task_group(python_callable=None, *tg_args, **tg_kwargs):
147     """
148     Python TaskGroup decorator.
149 
150     This wraps a function into an Airflow TaskGroup. When used as the
151     ``@task_group()`` form, all arguments are forwarded to the underlying
152     TaskGroup class. Can be used to parametrize TaskGroup.
153 
154     :param python_callable: Function to decorate.
155     :param tg_args: Positional arguments for the TaskGroup object.
156     :param tg_kwargs: Keyword arguments for the TaskGroup object.
157     """
158     if callable(python_callable):
159         return TaskGroupDecorator(function=python_callable, kwargs=tg_kwargs)
160     return cast("Callable[[T], T]", functools.partial(TaskGroupDecorator, kwargs=tg_kwargs))
161
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/decorators/task_group.py b/airflow/decorators/task_group.py
--- a/airflow/decorators/task_group.py
+++ b/airflow/decorators/task_group.py
@@ -42,7 +42,7 @@
 class TaskGroupDecorator(Generic[R]):
     """:meta private:"""
 
-    function: Callable[..., R] = attr.ib(validator=attr.validators.is_callable())
+    function: Callable[..., Optional[R]] = attr.ib(validator=attr.validators.is_callable())
     kwargs: Dict[str, Any] = attr.ib(factory=dict)
     """kwargs for the TaskGroup"""
 
@@ -62,9 +62,24 @@
         return TaskGroup(**kwargs)
 
     def __call__(self, *args, **kwargs) -> R:
-        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs):
+        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs) as task_group:
             # Invoke function to run Tasks inside the TaskGroup
-            return self.function(*args, **kwargs)
+            retval = self.function(*args, **kwargs)
+
+        # If the task-creating function returns a task, forward the return value
+        # so dependencies bind to it. This is equivalent to
+        #   with TaskGroup(...) as tg:
+        #       t2 = task_2(task_1())
+        #   start >> t2 >> end
+        if retval is not None:
+            return retval
+
+        # Otherwise return the task group as a whole, equivalent to
+        #   with TaskGroup(...) as tg:
+        #       task_1()
+        #       task_2()
+        #   start >> tg >> end
+        return task_group
 
     def partial(self, **kwargs) -> "MappedTaskGroupDecorator[R]":
         return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).partial(**kwargs)

{"golden_diff": "diff --git a/airflow/decorators/task_group.py b/airflow/decorators/task_group.py\n--- a/airflow/decorators/task_group.py\n+++ b/airflow/decorators/task_group.py\n@@ -42,7 +42,7 @@\n class TaskGroupDecorator(Generic[R]):\n     \"\"\":meta private:\"\"\"\n \n-    function: Callable[..., R] = attr.ib(validator=attr.validators.is_callable())\n+    function: Callable[..., Optional[R]] = attr.ib(validator=attr.validators.is_callable())\n     kwargs: Dict[str, Any] = attr.ib(factory=dict)\n     \"\"\"kwargs for the TaskGroup\"\"\"\n \n@@ -62,9 +62,24 @@\n         return TaskGroup(**kwargs)\n \n     def __call__(self, *args, **kwargs) -> R:\n-        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs):\n+        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs) as task_group:\n             # Invoke function to run Tasks inside the TaskGroup\n-            return self.function(*args, **kwargs)\n+            retval = self.function(*args, **kwargs)\n+\n+        # If the task-creating function returns a task, forward the return value\n+        # so dependencies bind to it. This is equivalent to\n+        #   with TaskGroup(...) as tg:\n+        #       t2 = task_2(task_1())\n+        #   start >> t2 >> end\n+        if retval is not None:\n+            return retval\n+\n+        # Otherwise return the task group as a whole, equivalent to\n+        #   with TaskGroup(...) as tg:\n+        #       task_1()\n+        #       task_2()\n+        #   start >> tg >> end\n+        return task_group\n \n     def partial(self, **kwargs) -> \"MappedTaskGroupDecorator[R]\":\n         return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).partial(**kwargs)\n", "issue": "Error setting dependencies on task_group defined using the decorator\n### Apache Airflow version\n\n2.2.2 (latest released)\n\n### Operating System\n\nMacOS 11.6.1\n\n### Versions of Apache Airflow Providers\n\n$ pip freeze | grep airflow\r\napache-airflow==2.2.2\r\napache-airflow-providers-celery==2.1.0\r\napache-airflow-providers-ftp==2.0.1\r\napache-airflow-providers-http==2.0.1\r\napache-airflow-providers-imap==2.0.1\r\napache-airflow-providers-sqlite==2.0.1\n\n### Deployment\n\nOther\n\n### Deployment details\n\n`airflow standalone`\n\n### What happened\n\n```\r\nAttributeError: 'NoneType' object has no attribute 'update_relative'\r\n```\n\n### What you expected to happen\n\nTask group should be set as downstream of `start` task, and upstream of `end` task\n\n### How to reproduce\n\n* Add the following code to dags folder\r\n\r\n```python\r\nfrom datetime import datetime\r\n\r\nfrom airflow.decorators import dag, task, task_group\r\n\r\n\r\n@dag(start_date=datetime(2023, 1, 1), schedule_interval=\"@once\")\r\ndef test_dag_1():\r\n    @task\r\n    def start():\r\n        pass\r\n\r\n    @task\r\n    def do_thing(x):\r\n        print(x)\r\n\r\n    @task_group\r\n    def do_all_things():\r\n        do_thing(1)\r\n        do_thing(2)\r\n\r\n    @task\r\n    def end():\r\n        pass\r\n\r\n    start() >> do_all_things() >> end()\r\n\r\n\r\ntest_dag_1()\r\n```\r\n\r\n* Run `airflow standalone`\n\n### Anything else\n\n_No response_\n\n### Are you willing to submit PR?\n\n- [ ] Yes I am willing to submit a PR!\n\n### Code of Conduct\n\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"\nA TaskGroup is a collection of closely related tasks on the same DAG that should be grouped\ntogether when the DAG is displayed graphically.\n\"\"\"\nimport functools\nimport warnings\nfrom inspect import signature\nfrom typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Optional, TypeVar, cast, overload\n\nimport attr\n\nfrom airflow.utils.task_group import MappedTaskGroup, TaskGroup\n\nif TYPE_CHECKING:\n    from airflow.models import DAG\n\nF = TypeVar(\"F\", bound=Callable[..., Any])\nT = TypeVar(\"T\", bound=Callable)\nR = TypeVar(\"R\")\n\ntask_group_sig = signature(TaskGroup.__init__)\n\n\n@attr.define\nclass TaskGroupDecorator(Generic[R]):\n    \"\"\":meta private:\"\"\"\n\n    function: Callable[..., R] = attr.ib(validator=attr.validators.is_callable())\n    kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"kwargs for the TaskGroup\"\"\"\n\n    @function.validator\n    def _validate_function(self, _, f):\n        if 'self' in signature(f).parameters:\n            raise TypeError('@task_group does not support methods')\n\n    @kwargs.validator\n    def _validate(self, _, kwargs):\n        task_group_sig.bind_partial(**kwargs)\n\n    def __attrs_post_init__(self):\n        self.kwargs.setdefault('group_id', self.function.__name__)\n\n    def _make_task_group(self, **kwargs) -> TaskGroup:\n        return TaskGroup(**kwargs)\n\n    def __call__(self, *args, **kwargs) -> R:\n        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs):\n            # Invoke function to run Tasks inside the TaskGroup\n            return self.function(*args, **kwargs)\n\n    def partial(self, **kwargs) -> \"MappedTaskGroupDecorator[R]\":\n        return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).partial(**kwargs)\n\n    def map(self, **kwargs) -> R:\n        return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).map(**kwargs)\n\n\n@attr.define\nclass MappedTaskGroupDecorator(TaskGroupDecorator[R]):\n    \"\"\":meta private:\"\"\"\n\n    partial_kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"static kwargs for the decorated function\"\"\"\n    mapped_kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"kwargs for the decorated function\"\"\"\n\n    def __call__(self, *args, **kwargs):\n        raise RuntimeError(\"A mapped @task_group cannot be called. Use `.map` and `.partial` instead\")\n\n    def _make_task_group(self, **kwargs) -> MappedTaskGroup:\n        tg = MappedTaskGroup(**kwargs)\n        tg.partial_kwargs = self.partial_kwargs\n        tg.mapped_kwargs = self.mapped_kwargs\n        return tg\n\n    def partial(self, **kwargs) -> \"MappedTaskGroupDecorator[R]\":\n        if self.partial_kwargs:\n            raise RuntimeError(\"Already a partial task group\")\n        self.partial_kwargs.update(kwargs)\n        return self\n\n    def map(self, **kwargs) -> R:\n        if self.mapped_kwargs:\n            raise RuntimeError(\"Already a mapped task group\")\n        self.mapped_kwargs = kwargs\n\n        call_kwargs = self.partial_kwargs.copy()\n        duplicated_keys = set(call_kwargs).intersection(kwargs)\n        if duplicated_keys:\n            raise RuntimeError(f\"Cannot map partial arguments: {', '.join(sorted(duplicated_keys))}\")\n        call_kwargs.update({k: object() for k in kwargs})\n\n        return super().__call__(**call_kwargs)\n\n    def __del__(self):\n        if not self.mapped_kwargs:\n            warnings.warn(f\"Partial task group {self.function.__name__} was never mapped!\")\n\n\n# This covers the @task_group() case. Annotations are copied from the TaskGroup\n# class, only providing a default to 'group_id' (this is optional for the\n# decorator and defaults to the decorated function's name). Please keep them in\n# sync with TaskGroup when you can! Note that since this is an overload, these\n# argument defaults aren't actually used at runtime--the real implementation\n# does not use them, and simply rely on TaskGroup's defaults, so it's not\n# disastrous if they go out of sync with TaskGroup.\n@overload\ndef task_group(\n    group_id: Optional[str] = None,\n    prefix_group_id: bool = True,\n    parent_group: Optional[\"TaskGroup\"] = None,\n    dag: Optional[\"DAG\"] = None,\n    default_args: Optional[Dict[str, Any]] = None,\n    tooltip: str = \"\",\n    ui_color: str = \"CornflowerBlue\",\n    ui_fgcolor: str = \"#000\",\n    add_suffix_on_collision: bool = False,\n) -> Callable[[F], F]:\n    ...\n\n\n# This covers the @task_group case (no parentheses).\n@overload\ndef task_group(python_callable: F) -> F:\n    ...\n\n\ndef task_group(python_callable=None, *tg_args, **tg_kwargs):\n    \"\"\"\n    Python TaskGroup decorator.\n\n    This wraps a function into an Airflow TaskGroup. When used as the\n    ``@task_group()`` form, all arguments are forwarded to the underlying\n    TaskGroup class. Can be used to parametrize TaskGroup.\n\n    :param python_callable: Function to decorate.\n    :param tg_args: Positional arguments for the TaskGroup object.\n    :param tg_kwargs: Keyword arguments for the TaskGroup object.\n    \"\"\"\n    if callable(python_callable):\n        return TaskGroupDecorator(function=python_callable, kwargs=tg_kwargs)\n    return cast(\"Callable[[T], T]\", functools.partial(TaskGroupDecorator, kwargs=tg_kwargs))\n", "path": "airflow/decorators/task_group.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"\nA TaskGroup is a collection of closely related tasks on the same DAG that should be grouped\ntogether when the DAG is displayed graphically.\n\"\"\"\nimport functools\nimport warnings\nfrom inspect import signature\nfrom typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Optional, TypeVar, cast, overload\n\nimport attr\n\nfrom airflow.utils.task_group import MappedTaskGroup, TaskGroup\n\nif TYPE_CHECKING:\n    from airflow.models import DAG\n\nF = TypeVar(\"F\", bound=Callable[..., Any])\nT = TypeVar(\"T\", bound=Callable)\nR = TypeVar(\"R\")\n\ntask_group_sig = signature(TaskGroup.__init__)\n\n\n@attr.define\nclass TaskGroupDecorator(Generic[R]):\n    \"\"\":meta private:\"\"\"\n\n    function: Callable[..., Optional[R]] = attr.ib(validator=attr.validators.is_callable())\n    kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"kwargs for the TaskGroup\"\"\"\n\n    @function.validator\n    def _validate_function(self, _, f):\n        if 'self' in signature(f).parameters:\n            raise TypeError('@task_group does not support methods')\n\n    @kwargs.validator\n    def _validate(self, _, kwargs):\n        task_group_sig.bind_partial(**kwargs)\n\n    def __attrs_post_init__(self):\n        self.kwargs.setdefault('group_id', self.function.__name__)\n\n    def _make_task_group(self, **kwargs) -> TaskGroup:\n        return TaskGroup(**kwargs)\n\n    def __call__(self, *args, **kwargs) -> R:\n        with self._make_task_group(add_suffix_on_collision=True, **self.kwargs) as task_group:\n            # Invoke function to run Tasks inside the TaskGroup\n            retval = self.function(*args, **kwargs)\n\n        # If the task-creating function returns a task, forward the return value\n        # so dependencies bind to it. This is equivalent to\n        #   with TaskGroup(...) as tg:\n        #       t2 = task_2(task_1())\n        #   start >> t2 >> end\n        if retval is not None:\n            return retval\n\n        # Otherwise return the task group as a whole, equivalent to\n        #   with TaskGroup(...) as tg:\n        #       task_1()\n        #       task_2()\n        #   start >> tg >> end\n        return task_group\n\n    def partial(self, **kwargs) -> \"MappedTaskGroupDecorator[R]\":\n        return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).partial(**kwargs)\n\n    def map(self, **kwargs) -> R:\n        return MappedTaskGroupDecorator(function=self.function, kwargs=self.kwargs).map(**kwargs)\n\n\n@attr.define\nclass MappedTaskGroupDecorator(TaskGroupDecorator[R]):\n    \"\"\":meta private:\"\"\"\n\n    partial_kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"static kwargs for the decorated function\"\"\"\n    mapped_kwargs: Dict[str, Any] = attr.ib(factory=dict)\n    \"\"\"kwargs for the decorated function\"\"\"\n\n    def __call__(self, *args, **kwargs):\n        raise RuntimeError(\"A mapped @task_group cannot be called. Use `.map` and `.partial` instead\")\n\n    def _make_task_group(self, **kwargs) -> MappedTaskGroup:\n        tg = MappedTaskGroup(**kwargs)\n        tg.partial_kwargs = self.partial_kwargs\n        tg.mapped_kwargs = self.mapped_kwargs\n        return tg\n\n    def partial(self, **kwargs) -> \"MappedTaskGroupDecorator[R]\":\n        if self.partial_kwargs:\n            raise RuntimeError(\"Already a partial task group\")\n        self.partial_kwargs.update(kwargs)\n        return self\n\n    def map(self, **kwargs) -> R:\n        if self.mapped_kwargs:\n            raise RuntimeError(\"Already a mapped task group\")\n        self.mapped_kwargs = kwargs\n\n        call_kwargs = self.partial_kwargs.copy()\n        duplicated_keys = set(call_kwargs).intersection(kwargs)\n        if duplicated_keys:\n            raise RuntimeError(f\"Cannot map partial arguments: {', '.join(sorted(duplicated_keys))}\")\n        call_kwargs.update({k: object() for k in kwargs})\n\n        return super().__call__(**call_kwargs)\n\n    def __del__(self):\n        if not self.mapped_kwargs:\n            warnings.warn(f\"Partial task group {self.function.__name__} was never mapped!\")\n\n\n# This covers the @task_group() case. Annotations are copied from the TaskGroup\n# class, only providing a default to 'group_id' (this is optional for the\n# decorator and defaults to the decorated function's name). Please keep them in\n# sync with TaskGroup when you can! Note that since this is an overload, these\n# argument defaults aren't actually used at runtime--the real implementation\n# does not use them, and simply rely on TaskGroup's defaults, so it's not\n# disastrous if they go out of sync with TaskGroup.\n@overload\ndef task_group(\n    group_id: Optional[str] = None,\n    prefix_group_id: bool = True,\n    parent_group: Optional[\"TaskGroup\"] = None,\n    dag: Optional[\"DAG\"] = None,\n    default_args: Optional[Dict[str, Any]] = None,\n    tooltip: str = \"\",\n    ui_color: str = \"CornflowerBlue\",\n    ui_fgcolor: str = \"#000\",\n    add_suffix_on_collision: bool = False,\n) -> Callable[[F], F]:\n    ...\n\n\n# This covers the @task_group case (no parentheses).\n@overload\ndef task_group(python_callable: F) -> F:\n    ...\n\n\ndef task_group(python_callable=None, *tg_args, **tg_kwargs):\n    \"\"\"\n    Python TaskGroup decorator.\n\n    This wraps a function into an Airflow TaskGroup. When used as the\n    ``@task_group()`` form, all arguments are forwarded to the underlying\n    TaskGroup class. Can be used to parametrize TaskGroup.\n\n    :param python_callable: Function to decorate.\n    :param tg_args: Positional arguments for the TaskGroup object.\n    :param tg_kwargs: Keyword arguments for the TaskGroup object.\n    \"\"\"\n    if callable(python_callable):\n        return TaskGroupDecorator(function=python_callable, kwargs=tg_kwargs)\n    return cast(\"Callable[[T], T]\", functools.partial(TaskGroupDecorator, kwargs=tg_kwargs))\n", "path": "airflow/decorators/task_group.py"}]}