from typing import List, Optional from pydantic import BaseModel from app.schemas.jm_homework_datasource_relation import JmHomeworkDatasourceRelationCreate class JmHomeworkBase(BaseModel): # 作业名称 name: str # 作业类型 type: str # 作业分类 tag: str # 镜像类别(0:默认,1:自建) image_type: Optional[int] # 执行镜像 image_url: str # DAG_ID dag_uuid: str # DAG文件地址 dag_url: str # 脚本文件 script_file: str # 执行命令 execute_command: Optional[str] class JmHomeworkCreate(JmHomeworkBase): relation_list: Optional[List[JmHomeworkDatasourceRelationCreate]] = None class Config: schema_extra = { "example": { "name": "监察流水线", "type": "java", "tag": "作业监测", "image_type": 0, "image_url": "/test/images/example", "dag_uuid": "", "dag_url": "", "script_file": "/test/scripts/example", "execute_command": "ls", "relation_list": [ { "type": "input", "datasource_id": 1, "node_uuid": "77a6e831-bb62-42e1-be8d-97699db00e73", "table": "jm_homework_datasource_relation", } ] } } class JmHomeworkUpdate(JmHomeworkBase): relation_list: Optional[List[JmHomeworkDatasourceRelationCreate]] = None class Config: schema_extra = { "example": { "name": "监察流水线", "type": "DAG", "tag": "作业监测", "image_type": 0, "image_url": "/test/images/example", "dag_uuid": "", "dag_url": "", "script_file": "/test/scripts/example", "execute_command": "ls", "relation_list": [ { "type": "input", "datasource_id": 1, "node_uuid": "77a6e831-bb62-42e1-be8d-97699db00e73", "table": "jm_homework_datasource_relation", } ] } } class JmHomework(JmHomeworkBase): id: int # 用户ID user_id: str # 项目ID project_id: str # 创建时间 create_time: int # 更新时间 update_time: int # 状态 status: int class Config: orm_mode = True