mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-20 20:39:55 +00:00
zhangtao
4b9ebfd4ed
refactor(前端): 重构任务管理界面为卡片式布局 fix(后端): 修复任务日志记录中的job_id类型问题 perf(调度器): 优化立即执行任务的日志记录机制 style(前端): 调整多个页面的高度计算方式
58 lines
2.2 KiB
Python
58 lines
2.2 KiB
Python
from fastapi import Query
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
Field,
|
|
)
|
|
|
|
from app.common.enums import QueueEnum
|
|
from app.core.base_schema import BaseSchema
|
|
|
|
|
|
class JobCreateSchema(BaseModel):
|
|
"""执行日志创建模型"""
|
|
|
|
job_id: str = Field(..., description="任务ID")
|
|
job_name: str | None = Field(default=None, description="任务名称")
|
|
trigger_type: str | None = Field(default=None, description="触发方式")
|
|
status: str = Field(default="pending", description="执行状态")
|
|
next_run_time: str | None = Field(default=None, description="下次执行时间")
|
|
job_state: str | None = Field(default=None, description="任务状态信息")
|
|
result: str | None = Field(default=None, description="执行结果")
|
|
error: str | None = Field(default=None, description="错误信息")
|
|
|
|
|
|
class JobUpdateSchema(BaseModel):
|
|
"""执行日志更新模型"""
|
|
|
|
status: str | None = Field(default=None, description="执行状态")
|
|
next_run_time: str | None = Field(default=None, description="下次执行时间")
|
|
job_state: str | None = Field(default=None, description="任务状态信息")
|
|
result: str | None = Field(default=None, description="执行结果")
|
|
error: str | None = Field(default=None, description="错误信息")
|
|
|
|
|
|
class JobOutSchema(JobCreateSchema, BaseSchema):
|
|
"""执行日志响应模型"""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
...
|
|
|
|
|
|
class JobQueryParam:
|
|
"""执行日志查询参数"""
|
|
|
|
def __init__(
|
|
self,
|
|
job_id: str | None = Query(None, description="任务ID"),
|
|
job_name: str | None = Query(None, description="任务名称"),
|
|
status: str | None = Query(None, description="执行状态"),
|
|
trigger_type: str | None = Query(None, description="触发方式"),
|
|
) -> None:
|
|
# 确保 job_id 是字符串类型
|
|
self.job_id = (QueueEnum.eq.value, str(job_id) if job_id is not None else None)
|
|
# 只有当 job_name 不为空时才添加查询条件
|
|
self.job_name = (QueueEnum.like.value, job_name) if job_name else None
|
|
self.status = (QueueEnum.eq.value, status)
|
|
self.trigger_type = (QueueEnum.eq.value, trigger_type)
|