Dynamic targets¶
Most transitions name one target. Two forms compute it instead, and the whole difference between them is when.
State.RETURN_VALUE |
State.GET_STATE |
|
|---|---|---|
| Target comes from | The method's return value | A function of the call's arguments |
| Resolved | After the body runs | Before the body runs |
| The body observes | The old state | The new state |
| An invalid target means | The body already ran | The body never runs |
| Declared targets | RETURN_VALUE(*states) |
GET_STATE(func, states=[...]) |
Pick on that last row. If deciding requires doing the work, you need
RETURN_VALUE. If the decision is knowable from the arguments, GET_STATE is
safer, because a bad target costs nothing.
class Comment(StateEngine):
state = State(ModerationState, default=ModerationState.NEW)
def __init__(self, text: str, is_public: bool = True) -> None:
self.text = text
self.is_public = is_public
self.state_seen_by_body: ModerationState | None = None
@state.transition(
source=ModerationState.NEW,
target=State.RETURN_VALUE(ModerationState.PUBLISHED, ModerationState.FOR_MODERATORS),
)
def publish(self) -> ModerationState:
"""Resolved after the body runs, from what the body returns."""
self.state_seen_by_body = self.state
return ModerationState.PUBLISHED if self.is_public else ModerationState.FOR_MODERATORS
@state.transition(
source=ModerationState.NEW,
target=State.GET_STATE(
decide_review_target,
states=[ModerationState.APPROVED, ModerationState.REJECTED],
),
)
def review(self, approved: bool) -> None:
"""Resolved before the body runs, from the call's arguments."""
self.state_seen_by_body = self.state
State.RETURN_VALUE¶
The target is whatever the method returns:
>>> comment = dynamic_target.Comment('Looks good to me')
>>> comment.publish()
<ModerationState.PUBLISHED: 'PUBLISHED'>
>>> comment.state
<ModerationState.PUBLISHED: 'PUBLISHED'>
The return value is passed through to the caller as well as consumed as the target, so nothing is swallowed.
Because it can only be known once the body has returned, the body still sees the state it started in:
A different instance takes the other branch, with no change to the declaration:
>>> flagged = dynamic_target.Comment('buy cheap watches', is_public=False)
>>> flagged.publish()
<ModerationState.FOR_MODERATORS: 'FOR_MODERATORS'>
>>> flagged.state
<ModerationState.FOR_MODERATORS: 'FOR_MODERATORS'>
The allowed states¶
The states passed to RETURN_VALUE(...) are enforced. A return value outside
them raises rather than landing on an undeclared state:
>>> class Review(StateEngine):
... state = State(['new', 'approved', 'rejected'], default='new')
... @state.transition(source='new', target=State.RETURN_VALUE('approved', 'rejected'))
... def decide(self):
... return 'deleted'
>>> review = Review()
>>> review.decide()
Traceback (most recent call last):
...
open_fsm.base.InvalidTargetState: 'Decide' resolved to 'deleted', which is not in the allowed target states ['approved', 'rejected']
The state is left untouched:
The body has already run by the time this raises
RETURN_VALUE validates after the fact. Whatever the method did before
returning — a charge taken, a mail sent — has happened, and only the state
is rolled back. That is the cost of deciding the target from the work
itself. If validation-before-work matters, use GET_STATE.
Omit the states entirely and any return value is accepted:
>>> class Freeform(StateEngine):
... state = State(['new'], default='new')
... @state.transition(source='new', target=State.RETURN_VALUE())
... def move(self, destination):
... return destination
>>> freeform = Freeform()
>>> freeform.move('somewhere-new')
'somewhere-new'
>>> freeform.state
'somewhere-new'
Convenient, and it gives up the one guarantee this form still offered. Declare the states unless you genuinely cannot enumerate them.
State.GET_STATE¶
The target is computed by a function receiving the instance and the call's own arguments:
def decide_review_target(comment: Comment, approved: bool) -> ModerationState:
"""Where ``Comment.review()`` lands, computed from the call's own argument."""
return ModerationState.APPROVED if approved else ModerationState.REJECTED
>>> approved = dynamic_target.Comment('Reasonable')
>>> approved.review(approved=True)
>>> approved.state
<ModerationState.APPROVED: 'APPROVED'>
>>> rejected = dynamic_target.Comment('Unreasonable')
>>> rejected.review(approved=False)
>>> rejected.state
<ModerationState.REJECTED: 'REJECTED'>
The function is called with exactly what the transition was called with, so its signature mirrors the method's. It runs before the body, which means the body already observes the state it is landing in:
An invalid target costs nothing¶
Because resolution happens first, a target outside states=[...] stops the call
before your method is entered:
>>> attempts = []
>>> def route(order, express):
... return 'teleported' if express else 'shipped'
>>> class Order(StateEngine):
... state = State(['paid', 'shipped'], default='paid')
... @state.transition(source='paid', target=State.GET_STATE(route, states=['shipped']))
... def dispatch(self, express):
... attempts.append(express)
>>> order = Order()
>>> order.dispatch(express=True)
Traceback (most recent call last):
...
open_fsm.base.InvalidTargetState: 'Dispatch' resolved to 'teleported', which is not in the allowed target states ['shipped']
>>> attempts
[]
>>> order.state
'paid'
Nothing was appended, because the body was never reached.
Handling the failure¶
InvalidTargetState is a TransitionNotAllowed, so an existing handler already
catches it, and it carries the detail separately:
>>> try:
... review.decide()
... except InvalidTargetState as error:
... print(error.target)
... print(error.allowed_states)
deleted
('approved', 'rejected')
RETURN_VALUE keeps a tuple, GET_STATE keeps a list
error.allowed_states is whichever the transition declared —
RETURN_VALUE(*states) collects varargs into a tuple, GET_STATE(states=[...])
stores the list you passed. Compare with in, or normalise with list(),
rather than comparing the container itself.
How they appear in listings¶
A declared target is a real edge, so introspection reports it like any other:
>>> transition = list(dynamic_target.Comment.review.get_transitions())[0]
>>> transition.declared_targets()
[<ModerationState.APPROVED: 'APPROVED'>, <ModerationState.REJECTED: 'REJECTED'>]
An unrestricted form has nothing to declare, so it reports none:
This is what get_outgoing_transitions() uses to decide
whether a State.ANY transition is a self-loop. An unrestricted dynamic target
can never be ruled out, so it is always listed as outgoing — one more reason to
enumerate the states.
Related guides¶
- Transitions — the ordinary, declared target
- Introspection — what these look like from outside
- Errors —
InvalidTargetStatein full