如何在Pytransitions的AsyncMachine中实现异步依赖回调的顺序执行？

I get exactly what you're dealing with here—when using transitions.extensions.asyncio.AsyncMachine, the async callbacks specified in prepare, before, or after lists run concurrently by default. This breaks dependencies between callbacks, like how your initialize2 couldn't access the value set by initialize1 in time, leading to that None_3Done output instead of the expected 1Done_2Done_3Done.

Here are two clean solutions that keep your callbacks focused and reusable:

Solution 1: Custom AsyncMachine for Global Sequential Execution #

The default AsyncMachine uses asyncio.gather to run all async callbacks at once. We can override this behavior by creating a subclass that executes callbacks one after another:

import asyncio
from transitions.extensions.asyncio import AsyncMachine

class SequentialAsyncMachine(AsyncMachine):
    async def _process_callback(self, callbacks, event_data, *args, **kwargs):
        # Run each callback in order, waiting for the previous to finish
        for callback in self._get_callbacks(callbacks, event_data):
            await self._run_callback(callback, event_data, *args, **kwargs)

class Model:
    STATES = ['A', 'B']
    TRANSITIONS = [
        {'trigger': 'next', 'source': 'A', 'dest': 'B', 'prepare': ['initialize1', 'initialize2', 'initialize3'], 'after': ['show_attributes']}
    ]

    def __init__(self, name, state='initial'):
        self.name = name
        self.state = state
        self.attribute_1 = None
        self.attribute_2 = None
        self.attribute_3 = None

    async def initialize1(self):
        await asyncio.sleep(1)
        self.attribute_1 = '1Done'
        print(f'{self.name} {self.state} -> Initialized1: ', self.attribute_1)

    async def initialize2(self):
        await asyncio.sleep(0.5)
        self.attribute_2 = f'{self.attribute_1}_2Done'
        print(f'{self.name} {self.state} -> Initialized2: ', self.attribute_2)

    async def initialize3(self):
        self.attribute_3 = f'{self.attribute_2}_3Done'
        print(f'{self.name} {self.state} -> Initialized3: ', self.attribute_3)

    async def show_attributes(self):
        print(f'{self.name} {self.state} -> Showing all: {self.attribute_3}')

# Use our custom sequential machine instead of the default AsyncMachine
machine = SequentialAsyncMachine(
    model=None,
    states=Model.STATES,
    transitions=Model.TRANSITIONS,
    initial=None,
    queued='model'
)

async def main():
    model1 = Model(name='Model1', state='A')
    machine.add_model(model1, initial=model1.state)
    await machine.dispatch('next')

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

This change makes all your prepare, before, and after async callbacks run in the order they're listed, no need to modify your existing callback functions. It keeps each callback focused on its single task while ensuring dependencies are respected.

Solution 2: Targeted Sequential Wrapper (For Specific Transitions) #

If you only need sequential execution for specific callback groups (not globally), you can create a reusable wrapper function to chain callbacks:

def sequential(*funcs):
    async def wrapper(event_data):
        for func in funcs:
            await func(event_data.model)
    return wrapper

# Update your transitions to use the wrapper
TRANSITIONS = [
    {'trigger': 'next', 'source': 'A', 'dest': 'B', 
     'prepare': [sequential(Model.initialize1, Model.initialize2, Model.initialize3)], 
     'after': ['show_attributes']}
]

This is useful if you want to mix concurrent and sequential behavior across different transitions, but the global subclass approach is better for large codebases where you need consistent sequential execution everywhere.

Why Your Original Code Failed #

The default AsyncMachine spins up all three initialize tasks at the same time. initialize2 started running before initialize1 finished setting attribute_1, so it pulled a None value. This cascaded to initialize3, resulting in the broken None_3Done output.

内容的提问来源于stack exchange，提问作者Saurav Kumar