mnote/services/ingest_service/app/models/job.py

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional


@dataclass
class IngestJob:
    """表示一次 ingest 任务，后续可持久化到数据库"""

    id: str
    notebook_id: Optional[str]
    doc_id: Optional[str]
    status: str = "queued"
    source: str = "manual"
    blocks: List[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)
    error_message: Optional[str] = None
    payload: Dict[str, str] = field(default_factory=dict)

    def mark_running(self) -> None:
        self.status = "running"
        self.updated_at = datetime.utcnow()

    def mark_done(self) -> None:
        self.status = "succeeded"
        self.updated_at = datetime.utcnow()

    def mark_failed(self, message: str) -> None:
        self.status = "failed"
        self.error_message = message
        self.updated_at = datetime.utcnow()