33 lines
1004 B
Python
33 lines
1004 B
Python
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()
|