How to cancel an asynchronous transition with PyTransition

Question

I have a queued PyTransition state machine, with some state doing work in on_enter. However, I want the user to be able to stop the machine at any time without waiting. For that, I need a way to cancel transitions.

Here is what I've found for now. However, accessing the _transition_queue_dict seems like a hack. Is there a proper way?

#!/usr/bin/env python3

import asyncio
import logging

from transitions.extensions.asyncio import AsyncMachine

logging.getLogger('transitions').setLevel(logging.DEBUG)


class Model:
    async def do_long_work(self):
        print("Working...")
        await asyncio.sleep(10)
        print("Worked!")

    async def print_stop(self):
        print("Stopped!")

    async def interrupt(self):
        global machine
        await asyncio.sleep(1)
        for task in machine.async_tasks[id(self)]:
            task.cancel()
        machine._transition_queue_dict[id(model)].clear()

        await self.stop()

model = Model()
machine = AsyncMachine(model=model, queued=True)

machine.add_states('running', on_enter=[model.do_long_work])
machine.add_states('stopped', on_enter=[model.print_stop])
machine.add_transition('run', 'initial', 'running')
machine.add_transition('stop', 'running', 'stopped')


async def run():
    await asyncio.gather(machine.dispatch('run'), model.interrupt())

asyncio.run(run())

I use the last commit on master (3836dc4).

Answer 1

The problem here is that you pass queued=True which instruct a machine to put new events into model queues and process events sequentially. Since you want to be able to interrupt events, omitting queued=True or setting queued=False (default value) will cancel events when you transition away/exit a state. No need to tinker with internal queues in this case.

import asyncio
import logging

from transitions.extensions.asyncio import AsyncMachine

logging.getLogger('transitions').setLevel(logging.DEBUG)


class Model:
    async def do_long_work(self):
        print("Working...")
        await asyncio.sleep(10)
        print("Worked!")

    async def print_stop(self):
        print("Stopped!")

    async def interrupt(self):
        await asyncio.sleep(1)
        await self.stop()


model = Model()
machine = AsyncMachine(model=model, queued=False)

machine.add_states('running', on_enter=[model.do_long_work])
machine.add_states('stopped', on_enter=[model.print_stop])
machine.add_transition('run', 'initial', 'running')
machine.add_transition('stop', 'running', 'stopped')


async def run():
    await asyncio.gather(machine.dispatch('run'), model.interrupt())

asyncio.run(run())
# Working...
# Stopped!