Get task by ID
(self, task_id: str)
| 1165 | return cursor.lastrowid |
| 1166 | |
| 1167 | async def get_task(self, task_id: str) -> Optional[Task]: |
| 1168 | """Get task by ID""" |
| 1169 | async with self._connect() as db: |
| 1170 | db.row_factory = aiosqlite.Row |
| 1171 | cursor = await db.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)) |
| 1172 | row = await cursor.fetchone() |
| 1173 | if row: |
| 1174 | task_dict = dict(row) |
| 1175 | # Parse result_urls from JSON |
| 1176 | if task_dict.get("result_urls"): |
| 1177 | task_dict["result_urls"] = json.loads(task_dict["result_urls"]) |
| 1178 | return Task(**task_dict) |
| 1179 | return None |
| 1180 | |
| 1181 | async def update_task(self, task_id: str, **kwargs): |
| 1182 | """Update task""" |