mis98zb commented on code in PR #32122:
URL: https://github.com/apache/airflow/pull/32122#discussion_r1255443742
##########
airflow/decorators/task_group.py:
##########
@@ -124,14 +124,14 @@ def partial(self, **kwargs: Any) ->
_TaskGroupFactory[FParams, FReturn]:
# TODO: fixme when mypy gets compatible with new attrs
return attr.evolve(self, partial_kwargs=kwargs) # type:
ignore[arg-type]
- def expand(self, **kwargs: OperatorExpandArgument) -> DAGNode:
+ def expand(self, concurrency_limit: int = None, **kwargs:
OperatorExpandArgument) -> DAGNode:
Review Comment:
I have a question for decorated.
I predefined some common flow as task group:
```python
@task_group(group_id=f'runScopeJob')
def run_job(task_name: str, submit_param: dict, check_interval: int,
timeout: timedelta) -> None:
@task(task_id=f'submit{task_name}Job')
def submit_scope_job(job_param: dict, ti=None) -> str:
# do something....
return job_url
@task.sensor(task_id=f'wait{task_name}Job',
poke_interval=check_interval, timeout=timeout.total_seconds(),
mode="reschedule")
def wait_job(job_url: str, ti=None) -> PokeReturnValue:
# do something....
if status == 'ok':
return PokeReturnValue(is_done=False, xcom_value=status)
elif status == 'ng':
return PokeReturnValue(is_done=True, xcom_value=status)
elif status == 'fatal':
raise AirflowFailException(f'Job {job_url} failed with status
{status}')
wait_scope_job(submit_scope_job(submit_param))
```
And use it in may place as `run_job.expand(xxxx)`.
If I put `max_active_groups_per_dagrun` as a parameters of the decorated
function, what's the graceful way to pass the value to the decorator?
--
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]
