add job_log

app/crud/__init__.py

@@ -1,3 +1,4 @@
 from app.crud.job_jdbc_datasource import *
 from app.crud.job_info import *
-from app.crud.datax_json import *
+from app.crud.datax_json import *
+from app.crud.job_log import *

app/crud/job_log.py

@@ -0,0 +1,19 @@
+import time
+from typing import List
+from app import models, schemas
+from sqlalchemy.orm import Session
+
+def create_job_log(db: Session, item: schemas.JobLogCreate):
+    db_item = models.JobLog(**item.dict(), **{"trigger_code": 2, "handle_code": 2})
+    db.add(db_item)
+    db.commit()
+    db.refresh(db_item)
+    return db_item
+
+def get_job_logs(db: Session):
+    res: List[models.JobLog] = db.query(models.JobLog).all()
+    return res
+
+def get_job_log_once(db: Session, id: int):
+    res: models.JobLog = db.query(models.JobLog).filter(models.JobLog.id == id).first()
+    return res

app/models/__init__.py

@@ -1,2 +1,3 @@
 from app.models.job_jdbc_datasource import *
-from app.models.job_info import *
+from app.models.job_info import *
+from app.models.job_log import *

app/models/job_log.py

@@ -0,0 +1,38 @@
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
+
+from app.models.database import BaseModel
+
+class JobLog(BaseModel):
+    __tablename__ = "job_log"
+
+    id = Column(Integer, primary_key=True, index=True)
+    # 执行器主键ID
+    job_group = Column(Integer, nullable=False)
+    # 任务主键ID
+    job_id = Column(Integer, nullable=False)
+    # 执行器地址，本次执行的地址
+    executor_address = Column(String)
+    # 执行器任务handler
+    executor_handler = Column(String)
+    # 执行器任务参数
+    executor_param = Column(String)
+    # 执行器任务分片参数，格式如1/2
+    executor_sharding_param = Column(String)
+    # 失败重试次数
+    executor_fail_retry_count = Column(Integer)
+    # 调度时间
+    trigger_time = Column(Integer)
+    # 调度结果
+    trigger_code = Column(Integer, nullable=False)
+    # 调度日志
+    trigger_msg = Column(String)
+    # 执行时间
+    handle_time = Column(Integer)
+    # 执行状态
+    handle_code = Column(Integer, nullable=False)
+    # 执行日志
+    handle_msg = Column(String)
+    # 告警状态：0-默认、1-无需告警、2-告警成功、3-告警失败
+    alarm_status = Column(Integer, nullable=False)
+    # datax 进程ID
+    process_id = Column(String)

app/routers/job_log.py

@@ -0,0 +1,36 @@
+from typing import Optional
+from fastapi import APIRouter
+
+from fastapi import Depends
+from sqlalchemy.orm import Session
+from app import schemas
+
+import app.crud as crud
+from utils.sx_time import sxtimeit
+from utils.sx_web import web_try
+from fastapi_pagination import Page, add_pagination, paginate, Params
+
+from app import get_db
+
+router = APIRouter(
+    prefix="/jpt/joblog",
+    tags=["joblog-日志管理"],
+)
+
+@router.post("/")
+@web_try()
+@sxtimeit
+def create_job_log(item: schemas.JobLogCreate, db: Session = Depends(get_db)):
+    return crud.create_job_log(db, item)
+
+@router.get("/")
+@web_try()
+@sxtimeit
+def get_job_logs(params: Params=Depends(), db: Session = Depends(get_db)):
+    return paginate(crud.get_job_logs(db), params)
+
+@router.get("/getOnce/{id}")
+@web_try()
+@sxtimeit
+def get_job_logs_once(id: int, db: Session = Depends(get_db)):
+    return crud.get_job_log_once(db, id)

app/schemas/__init__.py

@@ -1,3 +1,4 @@
 from app.schemas.job_jdbc_datasouce import *
 from app.schemas.job_info import *
-from app.schemas.datax_json import *
+from app.schemas.datax_json import *
+from app.schemas.job_log import *

app/schemas/job_log.py

@@ -0,0 +1,93 @@
+from re import S
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+class JobLogBase(BaseModel):
+    # 执行器主键ID
+    job_group: int
+    # 任务主键ID
+    job_id: int
+    # 执行器地址，本次执行的地址
+    executor_address: Optional[str]
+    # 执行器任务handler
+    executor_handler: Optional[str]
+    # 执行器任务参数
+    executor_param: str
+    # 执行器任务分片参数，格式如1/2
+    executor_sharding_param: str
+    # 失败重试次数
+    executor_fail_retry_count: int
+    # 告警状态：0-默认、1-无需告警、2-告警成功、3-告警失败
+    alarm_status: int
+
+
+
+class JobLogCreate(JobLogBase):
+    class Config:
+        schema_extra = {
+            "example": {
+                "job_group": 1,
+                "job_id": 1,
+                "executor_address": "FIRST",
+                "executor_handler": "",
+                "executor_param": "",
+                "executor_sharding_param": "1/2",
+                "executor_fail_retry_count": 3,
+                "alarm_status": 0
+            }
+        }
+
+class JobLogTriggerUpdate():
+    # 调度时间
+    trigger_time: int
+    # 调度结果
+    trigger_code: int
+    # 调度日志
+    trigger_msg: str
+    class Config:
+        schema_extra = {
+            "example": {
+                "trigger_time": 1588888888,
+                "trigger_code": 1,
+                "trigger_msg": "",
+            }
+        }
+
+class JobLogHandleUpdate():
+    # 执行时间
+    handle_time: int
+    # 执行状态
+    handle_code: int
+    # 执行日志
+    handle_msg: str
+    class Config:
+        schema_extra = {
+            "example": {
+                "handle_time": 1588888888,
+                "handle_code": 1,
+                "handle_msg": "",
+            }
+        }
+
+
+
+class JobLog(JobLogBase):
+    id: int
+     # 上次调度时间
+    trigger_time: int
+    # 调度状态: 0-停止 1-运行
+    trigger_code: int
+    # 上次调度时间
+    trigger_msg: str
+    # 执行时间
+    handle_time: int
+    # 执行状态
+    handle_code: int
+    # 执行日志
+    handle_msg: str
+    # datax 进程ID
+    process_id: str
+
+    class Config:
+        orm_mode = True

data/data.sql

@@ -65,4 +65,31 @@ ADD COLUMN `tag` varchar(20) NULL COMMENT '标签: 0开发1测试2开发' AFTER
 -- Alter Table structure for job_info
 -- ----------------------------
 ALTER TABLE `job_info`
-ADD COLUMN `delete_status` tinyint(4) NOT NULL COMMENT '数据状态：0-无效，1-有效' AFTER `last_handle_code`;
+ADD COLUMN `delete_status` tinyint(4) NOT NULL COMMENT '数据状态：0-无效，1-有效' AFTER `last_handle_code`;
+
+
+-- ----------------------------
+-- Table structure for job_log
+-- ----------------------------
+DROP TABLE IF EXISTS `job_log`;
+CREATE TABLE `job_log` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `job_group` int(11) NOT NULL COMMENT '执行器主键ID',
+  `job_id` int(11) NOT NULL COMMENT '任务，主键ID',
+  `executor_address` varchar(255) DEFAULT NULL COMMENT '执行器地址，本次执行的地址',
+  `executor_handler` varchar(255) DEFAULT NULL COMMENT '执行器任务handler',
+  `executor_param` varchar(512) DEFAULT NULL COMMENT '执行器任务参数',
+  `executor_sharding_param` varchar(20) DEFAULT NULL COMMENT '执行器任务分片参数，格式如 1/2',
+  `executor_fail_retry_count` int(11) DEFAULT '0' COMMENT '失败重试次数',
+  `trigger_time` bigint(13) DEFAULT NULL COMMENT '调度-时间',
+  `trigger_code` int(11) NOT NULL COMMENT '调度-结果 0-失败，1-成功，2-无',
+  `trigger_msg` text COMMENT '调度-日志',
+  `handle_time` bigint(13) DEFAULT NULL COMMENT '执行-时间',
+  `handle_code` int(11) NOT NULL COMMENT '执行-状态 0-失败，1-成功，2-无',
+  `handle_msg` text COMMENT '执行-日志',
+  `alarm_status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '告警状态：0-默认、1-无需告警、2-告警成功、3-告警失败',
+  `process_id` varchar(20) DEFAULT NULL COMMENT 'datax进程Id',
+  PRIMARY KEY (`id`) USING BTREE,
+  KEY `I_trigger_time` (`trigger_time`) USING BTREE,
+  KEY `I_handle_code` (`handle_code`) USING BTREE
+) ENGINE=InnoDB AUTO_INCREMENT=1581 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC;

production.ini

@@ -1,12 +1,12 @@
-; [DATABASE]
-; user = root
-; pwd = happylay
-; db_name = datax_web_dev
-; host = 192.168.199.107
-; port = 10086
 [DATABASE]
-user = aihubtest
-pwd = q9WBYDynEy@jh#5N
-db_name = aihubtest_dag_admin_db
-host = 10.254.12.7
-port = 3306
+user = root
+pwd = happylay
+db_name = datax_web_dev
+host = 192.168.199.107
+port = 10086
+; [DATABASE]
+; user = aihubtest
+; pwd = q9WBYDynEy@jh#5N
+; db_name = aihubtest_dag_admin_db
+; host = 10.254.12.7
+; port = 3306

server.py

@@ -7,6 +7,7 @@ from app.models.database import  engine, Base
 import app.routers.job_jdbc_datasource as router_jjds
 import app.routers.constants as router_constants
 import app.routers.job_info as router_job_info
+import app.routers.job_log as router_job_log
 import app.routers.datax_json as router_datax
 
 Base.metadata.create_all(bind=engine)
@@ -26,6 +27,7 @@ app.add_middleware(
 app.include_router(router_jjds.router)
 app.include_router(router_constants.router)
 app.include_router(router_job_info.router)
+app.include_router(router_job_log.router)
 app.include_router(router_datax.router)