Question: How can i do an async fanout?

[ShravanSunder]

I want to implement a fanout/gather pattern similar to Inngest's fan-out jobs(https://www.inngest.com/docs/guides/fan-out-jobs) where I can:

1. Trigger multiple async operations in parallel from a single workflow
2. Collect all results when they complete

something like this the 2 examples beow

Example 1: Same operation for N items

@DBOS.workflow()
async def process_batch_workflow(items):
    # Fanout - process each item in parallel
    handles = []
    for item in items:
        handle = process_item.start_async(item)
        handles.append(handle)
    
    # Gather - collect all results
    results = await asyncio.gather(*[
        handle.get_result_async() for handle in handles
    ])
    
    return results

@DBOS.step()
async def process_item(item):
    # Process single item asynchronously
    async with aiohttp.ClientSession() as session:
        response = await session.post("https://api.example.com/process", json=item)
        return await response.json()

however each step can only run once right for example 1?

Example 2

@DBOS.workflow()
async def user_signup_workflow(user_data):
    # Fanout - trigger multiple async operations
    email_handle = send_welcome_email.start_async(user_data)
    stripe_handle = create_stripe_trial.start_async(user_data) 
    crm_handle = add_to_crm.start_async(user_data)
    
    # Gather - wait for all to complete
    results = await asyncio.gather(
        email_handle.get_result_async(),
        stripe_handle.get_result_async(), 
        crm_handle.get_result_async()
    )

   // do something with results in a step
    
    return ....

@DBOS.step()
async def send_welcome_email(user_data):
    # async email logic
    pass

@DBOS.step() 
async def create_stripe_trial(user_data):
    # async stripe logic
    pass

@DBOS.step()
async def add_to_crm(user_data):
    # async CRM logic  
    pass

Is there a cleaner async pattern for this?

[apoliakov]

Hey there! Thanks for your interest. You can use Queues.

Check out the doc here:
https://docs.dbos.dev/python/tutorials/queue-tutorial

There's a demo app here:
https://github.com/dbos-inc/dbos-demo-apps/tree/main/python/s3mirror

The pattern works for both sync and async. For the latter, make sure to use enqueue_async.

[apoliakov]

And if the queues are too heavyweight then a series of start_workflow_async calls can do it:
https://docs.dbos.dev/python/reference/contexts#start_workflow_async

[ShravanSunder]

i don't think that queues are the right absraction. how do you do a gather and the next step of the workflow? DBOS queues don't seem to have topics, so you can't also seggregate by workflow_id

[apoliakov]

It's true that DBOS queues don't have "topics" but you can create multiple queues and configure their concurrency as needed. The recommended pattern for gathering is like this:

queue = Queue("example_queue")

@DBOS.step()
async def process_task_async(task):
  ...

@DBOS.workflow()
async def process_tasks(tasks):
  task_handles = []
  for task in tasks:
    handle = await queue.enqueue_async(process_task_async, task)
    task_handles.append(handle)
  results = [await handle.get_result() for handle in task_handles]

The gather is not a "single step" but each get_result call is durable and acts as a mini-step.

To perform a non-blocking check on each task, you can call handle.get_status before deciding when to call handle.get_result.

To skip queues, you can also replace the call with enqueue_async to start_workflow_async in the above sequence. The difference is that queues allow you to impose concurrency controls and rate limits whereas start_workflow_async calls will execute as soon as possible.

Does this work for your needs?

[apoliakov]

Hi @ShravanSunder just wanted to check in. Was the answer above helpful?

[ShravanSunder]

i'll have to try it out with dbos paradigm.

But its not quite the fan out -> multi steps -> fan in that i wanted to do and possible to do in other durable workflow system.

[apoliakov]

Note that the enqueued items can be multi-step workflows, not just single steps. Is that what you mean?

[apoliakov]

Hey @ShravanSunder just wanted to check in and ask if this question is answered to your satisfaction? We'll close this if we don't hear back from you for another week or so.