mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-25 21:57:48 +00:00
zhangtao
d2a863652e
- Added a "Start Here" section in both English and Chinese README files to guide new users on running the project locally and exploring its features. - Introduced backend conventions for date and serialization handling, clarifying the use of Pydantic v2 and PostgreSQL. - Updated the "Quick Start" section with detailed steps for first-time local setup, including environment requirements and backend setup instructions. - Improved overall structure and clarity of documentation to facilitate better understanding for developers and contributors.
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)
|