MediumSchedulingPython 3
Workflow State Machine
Track dependent jobs through legal workflow states with retry and cancellation behavior.
40m3 sample tests8 hidden tests
Implement WorkflowRunner, an in-memory job tracker for dependent work. Jobs move through a small set of states:
text
1pending -> running -> succeeded
2pending -> running -> failed
3running -> retrying -> pending
4pending/running/retrying/failed -> canceledRequirements
add_job(job_id, deps=None, max_attempts=1)registers a job.add_jobraisesValueErroron duplicate ids,max_attempts < 1, or dependencies that are not already registered.ready()returns pending jobs whose dependencies have succeeded, in insertion order.transition(job_id, state, reason=None)validates state transitions and returns the final state.- Starting a job increments its attempt count.
transitiontorunningis allowed only when the job is currentlypendingand appears inready(); otherwise raiseValueError(message must include"dependencies"when the job is blocked on unsatisfied dependencies).- Other illegal edges raise
ValueErrorwhose message includes"invalid transition". - If a running job fails and still has attempts left, move it to
retrying; otherwise move it tofailed. Requesting"failed"may therefore return"retrying"when attempts remain. retryingjobs are not ready. The caller must explicitlytransition(..., "pending")to re-queue them.status(job_id)returns{"state": str, "attempts": int, "history": list[dict]}where each history entry is{"state", "reason", "attempt"}in transition order.attemptis the job's attempt count at the time of that transition.- Raise
ValueErrorfor invalid transitions or unknown jobs.
Example
python
1runner = WorkflowRunner()
2runner.add_job("extract")
3runner.add_job("train", deps=["extract"], max_attempts=2)
4
5assert runner.ready() == ["extract"]
6runner.transition("extract", "running")
7runner.transition("extract", "succeeded")
8assert runner.ready() == ["train"]
9
10# Multiple ready jobs stay in insertion order (not sorted by id).
11runner2 = WorkflowRunner()
12runner2.add_job("b")
13runner2.add_job("a")
14runner2.add_job("c", deps=["a"])
15assert runner2.ready() == ["b", "a"]Constraints
- Keep it single-process and in memory.
- Use deterministic ordering.
- Raise
ValueErrorfor invalid transitions or unknown jobs.
Editor