Bogdanp / Bogdanp/dramatiq

Feature request: pipeline with dynamic children

Aperta
#625 0 commenti 4 reazioni 0 assegnatari Vedi su GitHub
enhancement
Lingua principale
Python
Stelle
5.3k
Fork
383
Merge medio
9h 41m
PR unite (30g)
2

Descrizione

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.

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.