datax-admin/app/schemas/af_task.py

from typing import List, Dict, Optional
from pydantic import BaseModel


class AirflowTaskBase(BaseModel):
    name: str
    file_urls: Optional[List[str]] = []
    script: str
    cmd: Optional[str] = ""
    cmd_parameters: str
    envs: Optional[Dict[str, str]] = {}
    run_image: str


class AirflowTaskCreate(AirflowTaskBase):
    task_type: str
    user_id: int  # todo: 后续user id需要根据鉴权 在上写文自动获得


class AirflowTaskUpdate(AirflowTaskBase):
    pass
    # update_time: int


class AirflowTask(AirflowTaskBase):
    id: int
    task_type: str
    user_id: int
    create_time: int
    update_time: int

    class Config:
        orm_mode = True