import time from dataclasses import dataclass, field from typing import Any, Optional @dataclass class TaskRecord: task_id: str = "" status: str = "submitted" # submitted | processing | successes | failed message: str = "" type: str = "workflow" data: Optional[dict[str, Any]] = None request: Optional[dict[str, Any]] = None create_time: int = field(default_factory=lambda: int(time.time() * 1000)) update_time: int = field(default_factory=lambda: int(time.time() * 1000)) def touch(self): self.update_time = int(time.time() * 1000) def to_dict(self) -> dict: return { "task_id": self.task_id, "status": self.status, "message": self.message, "type": self.type, "data": self.data, "request": self.request, "create_time": self.create_time, "update_time": self.update_time, } @classmethod def from_dict(cls, d: dict) -> "TaskRecord": return cls( task_id=d.get("task_id", ""), status=d.get("status", "submitted"), message=d.get("message", ""), type=d.get("type", "workflow"), data=d.get("data"), request=d.get("request"), create_time=d.get("create_time", int(time.time() * 1000)), update_time=d.get("update_time", int(time.time() * 1000)), )