Snapshot of every registered task. Returns a list, not a view — callers iterating outside the lock should not see in-flight writes.
(self)
| 89 | return self._tasks.get(task_id) |
| 90 | |
| 91 | def all(self) -> list[TaskStateBase]: |
| 92 | """Snapshot of every registered task. Returns a list, not a view — |
| 93 | callers iterating outside the lock should not see in-flight writes.""" |
| 94 | with self._lock: |
| 95 | return list(self._tasks.values()) |
| 96 | |
| 97 | def by_type(self, task_type: TaskType) -> list[TaskStateBase]: |
| 98 | """Snapshot filtered to a single TaskType — used by deprecated |
no outgoing calls