ai-flow-python/app/repository/task_record.py

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)),
        )