Update task fields and persist. Accepts any Task field as kwarg.
(self, task_id: str, **kwargs)
| 77 | return task |
| 78 | |
| 79 | def update_task(self, task_id: str, **kwargs) -> Task: |
| 80 | """Update task fields and persist. Accepts any Task field as kwarg.""" |
| 81 | with self._lock: |
| 82 | task = self._tasks.get(task_id) |
| 83 | if not task: |
| 84 | raise ValueError(f"Task {task_id} not found") |
| 85 | for key, value in kwargs.items(): |
| 86 | if value is None: |
| 87 | continue |
| 88 | if key == "status" and isinstance(value, str): |
| 89 | value = TaskStatus(value) |
| 90 | if hasattr(task, key): |
| 91 | setattr(task, key, value) |
| 92 | task.updated_at = datetime.now(timezone.utc).isoformat() |
| 93 | self._revision_log.append( |
| 94 | { |
| 95 | "timestamp": task.updated_at, |
| 96 | "action": "updated", |
| 97 | "task_id": task_id, |
| 98 | "changes": {k: v for k, v in kwargs.items() if v is not None}, |
| 99 | } |
| 100 | ) |
| 101 | self._persist() |
| 102 | return task |
| 103 | |
| 104 | def get_task(self, task_id: str) -> Optional[Task]: |
| 105 | return self._tasks.get(task_id) |
no test coverage detected