Task Management
State machine for task lifecycle, observers for notifications, commands for undo. Three patterns that handle the three hardest parts of any project tracker.
Key Abstractions
Orchestrator that manages tasks, users, sprints, and coordinates notifications
Entity with state machine transitions: TODO to IN_PROGRESS to DONE
Can be assignee or reporter, receives notifications on subscribed tasks
Time-boxed container grouping tasks for a development iteration
Interface for receiving notifications when task state changes
Class Diagram
The Key Insight
A task management system is fundamentally a state machine with an audience. Tasks move through a defined lifecycle (TODO, IN_PROGRESS, DONE), and every transition matters to someone: the assignee, the reporter, the team lead, maybe a CI pipeline. You need the transitions to be strict and the notifications to be decoupled.
The state machine prevents chaos. Without explicit transition rules, someone will mark a task DONE that was never started, and your sprint metrics become meaningless. By encoding valid transitions in a map, the system enforces workflow discipline without relying on humans to follow process.
The Observer pattern handles the "audience" problem. When a task changes, you do not want the Task class importing an email library, a Slack SDK, and a webhook client. Observers let you bolt on notification channels without the task ever knowing they exist. The Command pattern rounds things out by making every action reversible. Accidentally moved a task to DONE? Undo it and the previous state is restored cleanly.
Requirements
Functional
- Create tasks with title, reporter, and initial TODO state
- Assign tasks to users
- Transition tasks through states: TODO -> IN_PROGRESS -> DONE (and reopen: DONE -> TODO)
- Group tasks into sprints
- Notify observers on state changes and assignments
- Undo the most recent action (transition or assignment)
Non-Functional
- Invalid state transitions must be rejected with clear error messages
- Adding new notification channels should not require modifying the Task class
- Undo history should be bounded in production to prevent unbounded memory growth
Design Decisions
Why a transition map instead of open-ended state changes?
Free-form transitions let tasks jump from TODO to DONE. That breaks workflow tracking, sprint burndown charts, and any process the team relies on. A transition map is a whitelist: if the move is not explicitly allowed, it does not happen. You can always add transitions later. Removing them from a permissive system is much harder.
Why Observer over direct notification calls?
If Task directly calls emailService.send() and slackService.post(), you have coupled your domain entity to infrastructure. Testing requires mocking two services. Adding a third channel means editing Task. With Observer, Task just says "something happened" and each observer decides what to do with that information. You can add, remove, or replace observers at runtime.
Why Command for undo instead of storing state snapshots?
Snapshots work for small objects. But tasks can accumulate comments, attachments, and history. Snapshotting all of that on every change is expensive. Commands only store the delta: what changed and what the previous value was. Undo is replaying that delta in reverse. Much lighter.
Why is Sprint just a container of task IDs?
A Sprint does not own tasks. Multiple sprints might reference the same task (moved from one sprint to the next). Storing IDs instead of task references means the Sprint is loosely coupled to the task lifecycle. Delete a sprint, the tasks survive.
Interview Follow-ups
- "How would you add subtasks?" Give Task a
parentIdfield and asubtaskslist. A parent task cannot move to DONE until all subtasks are DONE. Recursive state validation. - "How do you handle concurrent edits?" Optimistic locking with a version field on Task. If two people try to transition the same task, the second one gets a conflict error and must reload.
- "How would you add priority and due dates?" Add fields to Task and a comparator-based sorting strategy for the board view. Priority changes are just another Command type for undo support.
- "What about audit logging?" Add an AuditObserver that writes every event to an append-only log. Since it implements TaskObserver, the task itself never knows about auditing.
Code Implementation
1 from abc import ABC, abstractmethod
2 from enum import Enum
3 from datetime import datetime
4 import uuid
5
6
7 class TaskState(Enum):
8 TODO = "TODO"
9 IN_PROGRESS = "IN_PROGRESS"
10 DONE = "DONE"
11
12
13 # Valid transitions: which states can move to which
14 VALID_TRANSITIONS: dict[TaskState, set[TaskState]] = {
15 TaskState.TODO: {TaskState.IN_PROGRESS},
16 TaskState.IN_PROGRESS: {TaskState.DONE, TaskState.TODO},
17 TaskState.DONE: {TaskState.TODO}, # reopen
18 }
19
20
21 class User:
22 def __init__(self, user_id: str, name: str, email: str = ""):
23 self.id = user_id
24 self.name = name
25 self.email = email
26
27 def __repr__(self) -> str:
28 return f"User({self.name})"
29
30
31 class TaskObserver(ABC):
32 """Observer interface for task change notifications."""
33
34 @abstractmethod
35 def on_task_updated(self, task: "Task", event: str) -> None: ...
36
37
38 class ConsoleNotifier(TaskObserver):
39 def on_task_updated(self, task: "Task", event: str) -> None:
40 print(f" [Notification] Task '{task.title}' ({task.id[:8]}): {event}")
41
42
43 class Task:
44 """Entity with state machine and observer support."""
45
46 def __init__(self, title: str, reporter: User):
47 self.id = str(uuid.uuid4())
48 self.title = title
49 self.state = TaskState.TODO
50 self.reporter = reporter
51 self.assignee: User | None = None
52 self.created_at = datetime.now()
53 self._observers: list[TaskObserver] = []
54
55 def add_observer(self, observer: TaskObserver) -> None:
56 self._observers.append(observer)
57
58 def notify_observers(self, event: str) -> None:
59 for obs in self._observers:
60 obs.on_task_updated(self, event)
61
62 def transition(self, new_state: TaskState) -> None:
63 if new_state not in VALID_TRANSITIONS.get(self.state, set()):
64 raise ValueError(
65 f"Invalid transition: {self.state.value} -> {new_state.value}"
66 )
67 old_state = self.state
68 self.state = new_state
69 self.notify_observers(f"State changed: {old_state.value} -> {new_state.value}")
70
71 def assign(self, user: User) -> None:
72 self.assignee = user
73 self.notify_observers(f"Assigned to {user.name}")
74
75 def __repr__(self) -> str:
76 assignee_name = self.assignee.name if self.assignee else "Unassigned"
77 return f"Task({self.title}, {self.state.value}, {assignee_name})"
78
79
80 class Command(ABC):
81 """Command interface for undoable operations."""
82
83 @abstractmethod
84 def execute(self) -> None: ...
85
86 @abstractmethod
87 def undo(self) -> None: ...
88
89
90 class TransitionCommand(Command):
91 def __init__(self, task: Task, new_state: TaskState):
92 self._task = task
93 self._new_state = new_state
94 self._previous_state: TaskState | None = None
95
96 def execute(self) -> None:
97 self._previous_state = self._task.state
98 self._task.transition(self._new_state)
99
100 def undo(self) -> None:
101 if self._previous_state is None:
102 raise ValueError("Cannot undo: command was never executed")
103 old = self._task.state
104 self._task.state = self._previous_state
105 self._task.notify_observers(
106 f"Undo: reverted {old.value} -> {self._previous_state.value}"
107 )
108
109
110 class AssignCommand(Command):
111 def __init__(self, task: Task, user: User):
112 self._task = task
113 self._user = user
114 self._previous_assignee: User | None = None
115
116 def execute(self) -> None:
117 self._previous_assignee = self._task.assignee
118 self._task.assign(self._user)
119
120 def undo(self) -> None:
121 old_name = self._task.assignee.name if self._task.assignee else "None"
122 self._task.assignee = self._previous_assignee
123 new_name = self._previous_assignee.name if self._previous_assignee else "None"
124 self._task.notify_observers(f"Undo assign: {old_name} -> {new_name}")
125
126
127 class Sprint:
128 """Time-boxed container for tasks."""
129
130 def __init__(self, name: str):
131 self.name = name
132 self.task_ids: list[str] = []
133
134 def add_task(self, task_id: str) -> None:
135 if task_id not in self.task_ids:
136 self.task_ids.append(task_id)
137
138 def remove_task(self, task_id: str) -> None:
139 self.task_ids.remove(task_id)
140
141 def __repr__(self) -> str:
142 return f"Sprint({self.name}, {len(self.task_ids)} tasks)"
143
144
145 class TaskBoard:
146 """Orchestrates tasks, users, sprints, and undo history."""
147
148 def __init__(self):
149 self._tasks: dict[str, Task] = {}
150 self._users: dict[str, User] = {}
151 self._sprints: list[Sprint] = []
152 self._history: list[Command] = []
153 self._notifier = ConsoleNotifier()
154
155 def add_user(self, user: User) -> None:
156 self._users[user.id] = user
157
158 def create_task(self, title: str, reporter: User) -> Task:
159 task = Task(title, reporter)
160 task.add_observer(self._notifier)
161 self._tasks[task.id] = task
162 print(f" Created task: '{title}' (id: {task.id[:8]})")
163 return task
164
165 def move_task(self, task_id: str, new_state: TaskState) -> None:
166 task = self._tasks[task_id]
167 cmd = TransitionCommand(task, new_state)
168 cmd.execute()
169 self._history.append(cmd)
170
171 def assign_task(self, task_id: str, user: User) -> None:
172 task = self._tasks[task_id]
173 cmd = AssignCommand(task, user)
174 cmd.execute()
175 self._history.append(cmd)
176
177 def undo(self) -> None:
178 if not self._history:
179 print(" Nothing to undo.")
180 return
181 cmd = self._history.pop()
182 cmd.undo()
183
184 def create_sprint(self, name: str) -> Sprint:
185 sprint = Sprint(name)
186 self._sprints.append(sprint)
187 return sprint
188
189 def get_task(self, task_id: str) -> Task:
190 return self._tasks[task_id]
191
192
193 if __name__ == "__main__":
194 board = TaskBoard()
195
196 alice = User("u1", "Alice", "alice@example.com")
197 bob = User("u2", "Bob", "bob@example.com")
198 board.add_user(alice)
199 board.add_user(bob)
200
201 # Create tasks
202 print("--- Creating Tasks ---")
203 t1 = board.create_task("Implement login API", alice)
204 t2 = board.create_task("Write unit tests", alice)
205 t3 = board.create_task("Deploy to staging", bob)
206
207 # Assign and transition
208 print("\n--- Assign and Move ---")
209 board.assign_task(t1.id, bob)
210 board.move_task(t1.id, TaskState.IN_PROGRESS)
211 board.move_task(t1.id, TaskState.DONE)
212
213 # Sprint management
214 print("\n--- Sprint ---")
215 sprint = board.create_sprint("Sprint 1")
216 sprint.add_task(t1.id)
217 sprint.add_task(t2.id)
218 print(f" {sprint}")
219
220 # Undo the last transition
221 print("\n--- Undo ---")
222 board.undo()
223 print(f" Task state after undo: {board.get_task(t1.id).state.value}")
224
225 # Invalid transition
226 print("\n--- Invalid Transition ---")
227 try:
228 board.move_task(t2.id, TaskState.DONE) # Can't skip IN_PROGRESS
229 except ValueError as e:
230 print(f" Caught expected error: {e}")
231
232 # Move t2 properly
233 print("\n--- Valid Flow ---")
234 board.move_task(t2.id, TaskState.IN_PROGRESS)
235 board.move_task(t2.id, TaskState.DONE)
236
237 print(f"\n Final state t1: {board.get_task(t1.id)}")
238 print(f" Final state t2: {board.get_task(t2.id)}")
239 print("All scenarios completed.")Common Mistakes
- ✗Allowing arbitrary state transitions. A task jumping from TODO directly to DONE skips important workflow steps.
- ✗Notifying inside the Task class. That couples Task to every notification channel you will ever support.
- ✗Not storing the previous state in commands. Without it, undo has no idea what to revert to.
- ✗Making Sprint responsible for task state. Sprint groups tasks by time. State transitions are the task's own business.
Key Points
- ✓Task state transitions are explicit: TODO -> IN_PROGRESS -> DONE, with DONE -> TODO for reopening.
- ✓Observer pattern decouples notification logic from task state changes. Add email, Slack, or webhooks without touching Task.
- ✓Command pattern wraps state changes so you can undo them. Undo is just reversing the last command.
- ✓Sprints are containers, not owners. A task can move between sprints without losing its history.