How to implement a Celery-style chord

[anthonyatp]

Hi Taskiq team, thanks for the great project.

I’m using Taskiq as an async-friendly alternative to Celery and I’d love guidance on the idiomatic way to implement a Celery-style chord: trigger N tasks in parallel and then run a single callback only after all have finished.

Environment: RedisStreamBroker + RedisAsyncResultBackend (long TTL), FastAPI integration via taskiq_fastapi.init. I want clear behavior for retries, failures, and timeouts, and a straightforward way to aggregate results.
Here’s intent-only pseudo-code to illustrate the goal:

@broker.task
async def process_item(item_id: int) -> dict:
    ...

@broker.task
async def finalize() -> None:
    ...

async def orchestrate(items: list[int]) -> None:
    chord = (
        Chord(
            broker,
            process_item(i) for i in items,  # Create a task per item
        )
        .on_complete(finalize)
    )

    task = await chord.kiq()

I have looked into taskiq-pipelines but it's unclear how I would implement something like this. The new API design for pipeline looks promising but I was wondering if there was a more short-term solution you would recommend?

Thanks