Skip to content

SQLAlchemy Planned

Planned — not implemented yet

open_fsm.contrib.sqlalchemy does not exist. This page describes the API being designed for it, so the shape can be reviewed before it ships. Nothing on this page is executable, and no code block here is run by the test suite.

For what works today, see Binding state to storage.

Same shape, no saving

Like the Django integration, this is a flow class wrapping a mapped instance — the pattern Viewflow uses and the one open-fsm inherited. The mapped class keeps a plain column and knows nothing about the workflow.

What is different is that SQLAlchemy's unit of work already does the persisting. Assigning a mapped attribute marks the instance dirty, and the session writes it on flush — so this integration should supply the getter and setter and nothing else. Adding a save hook would issue a second UPDATE and defeat the unit of work.

That makes it the smallest of the three:
__init__ storing the instance Provided by ModelFlow
Getter and setter Derived from the attribute name
Saving Nothing — the session owns it
Concurrency A version column or with_for_update(), both already SQLAlchemy's

Proposed API

models.py — proposed, not yet available
class Report(Base):
    __tablename__ = 'reports'

    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    state_field: Mapped[ReportState] = mapped_column(
        SQLEnum(ReportState),
        default=ReportState.NEW,
    )
flows.py — proposed, not yet available
from open_fsm.contrib.sqlalchemy import ModelFlow, ModelState


class ReportFlow(ModelFlow):
    state = ModelState(ReportState, default=ReportState.NEW, attribute='state_field')

    @state.transition(source=ReportState.NEW, target=ReportState.APPROVED)
    def approve(self):
        self.instance.approvals.append(Approval())

    @state.transition(source=ReportState.NEW, target=ReportState.REJECTED)
    def reject(self):
        ...
services.py — proposed, not yet available
def approve_report(session, pk):
    report = session.scalars(
        select(Report).where(Report.id == pk).with_for_update()
    ).one()

    ReportFlow(report).approve()
    session.commit()

    return report

One transaction. The state change and whatever the transition body did to the object graph are flushed together.

Under discussion

Opinions are welcome on the issue tracker.

  • Whether ModelState is needed at all. Because the session handles saving, a getter and setter over a mapped attribute are two plain lines — see below. ModelState saves a little boilerplate and adds a dependency. This may end up a documentation page and no module, which would be the right outcome.
  • AsyncSession. The same synchronous-transition, awaited-commit split as Tortoise. The transition itself needs no changes; only the caller awaits.
  • Optimistic concurrency. __mapper_args__ = {'version_id_col': ...} turns a lost update into a StaleDataError at flush. Whether the integration should translate that into a TransitionNotAllowed is open — arguably it should not, since they fail at different times and mean different things.
  • on_success semantics. Mapping it onto an after_flush event rather than firing it at transition time would make the hook mean "persisted" instead of "state changed". That is a better guarantee and a bigger change.

What to do today

Two methods over the mapped attribute, and no save hook:

flows.py — works today
class ReportFlow(StateEngine):
    state = State(ReportState, default=ReportState.NEW)

    def __init__(self, report):
        self.report = report

    @state.getter()
    def _get_state(self):
        return self.report.state_field

    @state.setter()
    def _set_state(self, value):
        self.report.state_field = value

    @state.transition(source=ReportState.NEW, target=ReportState.APPROVED)
    def approve(self):
        ...

Note the absence of @state.on_success(). Assigning self.report.state_field marks the instance dirty, and session.commit() writes it.