video_decode/models/decode_result_record.py

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 DecodeResultRecord(BaseModel):
    table_name:       Annotated[Optional[str], Field(description='表名', exclude=True)] = 'decode_result_record'

    task_id:          Annotated[str, Field(description='任务ID')]
    model:            Annotated[str, Field(description='模型名称')]
    task_params:      Annotated[Optional[str], Field(description='任务参数', default=None)]
    task_status:      Annotated[Optional[int], Field(description='任务状态', default=1)]  # 1: 进行中, 2: 已完成, 3: 失败
    task_result:      Annotated[Optional[str], Field(description='任务结果', default=None)]
    create_timestamp: 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]))