Concurrent tasks with dependencies

[kevinhikaruevans]

I'd like to run several tasks with specific dynamic sequences. The dependencies can change, depending on the argument to run_start(). For example, run_start('some_task') might have a sequence that looks like:

I think I can use TaskiqDepends here but I'm not sure if that's the proper way to do that. Is it possible to pass arguments to a dependency? And is it possible to change the dependencies dynamically? For example, maybe I have another task called my_other_task that only runs run_step(1) and run_step(2) concurrently without run_step(3). Is that something that I can change programatically?

I was also thinking that I could run all the steps with gather() then in each step, wait for the dependent tasks to finish before proceeding. Is that possible to do? I was thinking something similar to await asyncio.gather(*[event.wait() for dependent_asyncio_events]), but using task results?

[s3rius]

TaskiqDepends has support for kwargs, but it cannot be used for dynamic dependency generation and you cannot use task arguments as dependency argument. Maybe this can be a good feature. I created an issue, but it's hard implement, so I cannot provide you with due date.

For now, you can either create multiple tasks, or call your function at the beginning.

Here's simple example of using kwargs in TaskiqDepends. As you can see, it's only simple static values.

import asyncio
from taskiq import InMemoryBroker, TaskiqDepends

broker = InMemoryBroker()

async def dependency(sleep_time: float) -> float:
    await asyncio.sleep(sleep_time)
    return sleep_time

@broker.task
async def short_sleep(sleept: float = TaskiqDepends(dependency, kwargs={"sleep_time": 0.5})):
    print(f"short_sleep: {sleept}")

@broker.task
async def long_sleep(sleept: float = TaskiqDepends(dependency, kwargs={"sleep_time": 2.0})):
    print(f"long_sleep: {sleept}")


async def main():
    await short_sleep.kiq()
    t2 = await long_sleep.kiq()
    await t2.wait_result()


if __name__ == "__main__":
    asyncio.run(main())

Regarding asyncio.gather. We actually have a taskiq.gather function that does exactly this.

About dependencies and all, maybe, if you were thinking not about dependencies, but tasks, maybe https://github.com/taskiq-python/taskiq-pipelines might be a good fit in your situation.