How to safely handle exceptions in async context?

Python State Machine version: 2.5.0
Python version: 3.12
Operating System: Linux

Description

I ran into this issue: there may be multiple coroutines issuing events to the machine, yet if a coroutine triggers an invalid transition, a different coroutine may receive the exception instead.

Here is a simple example:

from statemachine import Event, State, StateMachine
import asyncio


class Test(StateMachine):
    INITIAL = State(initial=True)
    FINAL = State(final=True)

    noop = Event(INITIAL.to(FINAL))

    @noop.on
    async def do_nothing(self, name):
        await asyncio.sleep(10)
        print(f"Did nothing via {name}")


test = Test()


if __name__ == "__main__":

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()

    async def fn1():
        await test.noop("fn1")

# Try/Except block is not working here
    async def fn2():
        try:
            await test.noop("fn2")
        except Exception as e:
            print(e)

    loop.create_task(fn1())
    loop.create_task(fn2())
    loop.run_forever()

Output is:

   ➜ python3 test.py 
Did nothing via fn1
Task exception was never retrieved
future: <Task finished name='Task-1' coro=<fn1() done> 
exception=TransitionNotAllowed("Can't Noop when in Final.")>
Traceback (most recent call last):
  File "test.py", line 29, in fn1
    await test.noop("fn1")
  File "/python3.12/site-packages/statemachine/engines/async_.py", line 66, in processing_loop
    result = await self._trigger(trigger_data)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/python3.12/site-packages/statemachine/engines/async_.py", line 96, in _trigger
    raise TransitionNotAllowed(trigger_data.event, state)
statemachine.exceptions.TransitionNotAllowed: Can't Noop when in Final.

It is clear that the invalid transition is triggered in fn2 yet the exception is thrown in fn1

While this is just a proof of concept, in my original use-case there are some places where I know that triggering a transition will not result in an exception and some other places where I know a try/catch is needed. However, if the exception is thrown in an unpredictable place I would be forced to implement exception handling everywhere and this makes stacking events complex.

Is just ignoring events that didn't result in transitions a valid option for your use case?

If so, you can just create the SM instance with Test(allow_event_without_transition=True)

Ref: https://python-statemachine.readthedocs.io/en/latest/api.html#statemachine

Hi, thanks for the quick answer and suggestion. This seems to work well for invalid transitions. Unluckily this is not a viable option for me however, as I was using transition's validators to throw meaningful exceptions that need to be catched by the same coroutine issuing the event.

Concretely:

class Test(StateMachine):
    INITIAL = State(initial=True)
    FINAL = State()

    noop = Event(INITIAL.to(FINAL))
    noop2 = Event(INITIAL.to(FINAL) | FINAL.to.itself())

    @noop2.on
    @noop.on
    async def do_nothing(self, name):
        await asyncio.sleep(5)
        print(f"Did nothing via {name}")

    @noop2.validators
    def raise_exception(self):
        # Do some checks ...
        # If checks fail
        print("noop2 is not allowed", flush=True)
        raise ValueError("noop2 is not allowed")

test = Test(allow_event_without_transition=True)

try:
    loop = asyncio.get_running_loop()
except RuntimeError:
    loop = asyncio.new_event_loop()

async def fn1():
    await test.noop("fn1")

async def fn2():
    try:
        await test.noop2("fn2")
    except ValueError as e:
        # Some logic
        print(e)
loop.create_task(fn1())
loop.create_task(fn2())
loop.run_forever()

However validator exceptions too are not guaranteed to be thrown in the same coroutine so I guess i'll have to revaluate my application's flow

I'm afraid that currently this is not supported out of the box. The problem is that the event's queue is consumed by a "critical section" that only the first caller can enter, and then the queue is consumed to exhaustion. While the critical section is closed by the first caller, other callers don't block and just put the event into the queue returning immediately.

So that's why eventually the first caller gets the second event.

But our thread gives me an idea that I can work on.... I don't know if will work and probably will break the public API of the library, but looks like promising: For each caller that don't get the chance to enter the critical section, instead of returning None, I could return a Promisse-like object, like a asyncio.Future or concurrent.futures.Future. And when the event is consumed, set the future.set_result(result) or future.set_exception(exception).

Seems like a good idea. Let me know if help is wanted

Update for v3

In v3, the new StateChart base class changes the defaults in a way that addresses both scenarios from this issue:

Scenario 1: `TransitionNotAllowed` in the wrong coroutine

StateChart sets allow_event_without_transition = True by default. Events that don't match any valid transition are silently ignored instead of raising TransitionNotAllowed. So this scenario no longer causes an exception at all.

Scenario 2: Validator/callback exceptions in the wrong coroutine

StateChart sets error_on_execution = True by default. Runtime exceptions in callbacks (guards, validators, actions, entry/exit) are caught by the engine and turned into an internal error.execution event — they never propagate to any caller. You can then handle errors declaratively within the statechart:

from statemachine import State, StateChart

class Test(StateChart):
    INITIAL = State(initial=True)
    FINAL = State()
    error_state = State(final=True)

    noop = INITIAL.to(FINAL, on="do_nothing")
    noop2 = INITIAL.to(FINAL, on="do_nothing", cond="check_allowed") | FINAL.to.itself(on="do_nothing")

    # Handle error.execution transitions
    error_execution = INITIAL.to(error_state) | FINAL.to(error_state)

    async def do_nothing(self, name):
        await asyncio.sleep(5)
        print(f"Did nothing via {name}")

    def check_allowed(self):
        raise ValueError("noop2 is not allowed")

    def on_enter_error_state(self, error=None, **kwargs):
        print(f"Caught error: {error}")

This follows the SCXML error handling specification — errors are handled as events within the statechart rather than as exceptions propagating to callers.

The underlying architecture

The fundamental design remains the same: the processing loop uses a non-blocking lock, so only one coroutine runs the loop while others enqueue and return immediately. This is the Run-to-Completion (RTC) model — each macrostep must complete atomically before the next event is processed.

A Future-based approach (where each caller gets a future that resolves with its event's result/exception) would allow per-caller error handling, but it's a significant architectural change. For now, the v3 approach of handling errors declaratively via error.execution transitions is the recommended pattern.

See the error handling docs and the processing model docs for more details.

feat: add future-based result routing for async concurrent events (#509)

test: add regression test for issue #509 example

This is now fixed via #567 with future-based result routing. Each concurrent caller gets its own event's result or exception — they no longer leak across coroutines.

Here's the original example from this thread, that now works for v3 (I've add a regression test at tests/testcases/test_issue509.py):

import asyncio

from statemachine import State, StateChart


class Test(StateChart):
    error_on_execution = False

    INITIAL = State(initial=True)
    FINAL = State()

    noop = INITIAL.to(FINAL, on="do_nothing")
    noop2 = INITIAL.to(FINAL, on="do_nothing", validators="raise_exception") | FINAL.to.itself(
        on="do_nothing", validators="raise_exception"
    )

    async def do_nothing(self, name):
        await asyncio.sleep(0.1)
        return f"Did nothing via {name}"

    def raise_exception(self):
        raise ValueError("noop2 is not allowed")


async def main():
    test = Test()
    await test.activate_initial_state()

    async def fn1():
        result = await test.send("noop", "fn1")
        print(f"fn1 got: {result}")  # "Did nothing via fn1"

    async def fn2():
        try:
            await test.send("noop2", "fn2")
        except ValueError as e:
            print(f"fn2 caught: {e}")  # "noop2 is not allowed"

    await asyncio.gather(
        asyncio.create_task(fn1()),
        asyncio.create_task(fn2()),
    )

asyncio.run(main())

Now fn1 receives its own result and fn2 catches its own ValueError — regardless of which coroutine holds the processing lock.

feat: future-based result routing for async concurrent events (#567)

Navigation Menu

Repository navigation

How to safely handle exceptions in async context? #509

Description

Update for v3

Scenario 1: `TransitionNotAllowed` in the wrong coroutine

Scenario 2: Validator/callback exceptions in the wrong coroutine

The underlying architecture

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

How to safely handle exceptions in async context? #509

Description

Description

Activity

ale18V commented on Dec 10, 2024

fgmacedo commented on Dec 10, 2024

ale18V commented on Dec 11, 2024

fgmacedo commented on Dec 11, 2024

ale18V commented on Dec 11, 2024

fgmacedo commented on Feb 15, 2026

Update for v3

Scenario 1: TransitionNotAllowed in the wrong coroutine

Scenario 2: Validator/callback exceptions in the wrong coroutine

The underlying architecture

fgmacedo commented on Feb 15, 2026

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions

Scenario 1: `TransitionNotAllowed` in the wrong coroutine