| 1234567891011121314151617181920212223242526272829303132 |
- from typing import Optional
- from pydantic import BaseModel, Field
- from typing_extensions import Annotated
- from utils.general import get_now_ts
- from utils.sync_mysql_help import mysql
- class EvaluateRecord(BaseModel):
- table_name: Annotated[Optional[str], Field(description='表名', exclude=True)] = 'evaluate_record'
-
- evaluate_id: Annotated[str, Field(description='任务ID', default=None)]
- task_id: Annotated[Optional[str], Field(description='任务ID', default=None)]
- status: Annotated[Optional[int], Field(description='任务状态', default=1)] # 1: 进行中, 2: 已完成, 3: 失败
- search_result: Annotated[Optional[str], Field(description='关键词结果', default=None)]
- evaluate_result: Annotated[Optional[str], Field(description='评估结果', default=None)]
- error_reason: Annotated[Optional[str], Field(description='失败原因', default=None)]
- created_at: Annotated[Optional[int], Field(description='任务创建时间戳', default_factory=get_now_ts)]
- def save(self):
- record = self.model_dump(exclude={'table_name'})
- keys = record.keys()
- sql = f'''INSERT INTO {self.table_name} ({", ".join(keys)}) VALUES ({", ".join(["%s"] * len(keys))})'''
- mysql.execute(sql, tuple([record[key] for key in keys]))
- async def async_save(self):
- record = self.model_dump(exclude={'table_name'})
- keys = record.keys()
- sql = f'''INSERT INTO {self.table_name} ({", ".join(keys)}) VALUES ({", ".join(["%s"] * len(keys))})'''
- await mysql.execute(sql, tuple([record[key] for key in keys]))
|
