Bogdanp / Bogdanp/dramatiq

Feature request: pipeline with dynamic children

Open
#625 0 comments 4 reactions 0 assignees View on GitHub
enhancement
Dominant language
Python
Stars
5.3k
Forks
383
Avg merge
9h 41m
Merged PRs (30d)
2

Description

I'm trying to migrate away from Celery, but I have the following scenario Workflow:
* Build a list of tasks
* process tasks (fanout)
* Join the results

In celery, I can do the following:

```python

@shared_task
def build_task_list(num: int) -> list[str]:
return [f"Task-{i}" for i in range(num)]

@shared_task
def process_and_gather(task_list: list[str], gather_task: Singature) -> None
return (group(process.s(t) for t in task_list) | gather_task).delay()

@shared_task
def process(task: str) -> str:
return f"{task}: done"

@shared_task
def combine_results(results: list[str]) -> str:
return '\n'.join(results)

# full workflow

workflow = build_task_list.s(100) | process_and_gather.s(gather_task=combine_results.s())
result = workflow.delay()
print(result.get())

```

It's not possible to do right now in dramatic, I believe it would be a great addition to the library.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.