This commit is contained in:
Marsway 2026-01-13 09:58:34 +08:00
parent a98697e376
commit 1435653b0b
3 changed files with 21 additions and 109 deletions

View File

@ -1,57 +0,0 @@
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
class RecentDateTimeFilter:
"""
最近时间范围筛选用于 started_at/created_at DateTime
- all / 1h / 24h / 7d / 30d
SQLAdmin 自定义 ColumnFilter 约定
- 必须提供 title / parameter_name
- 必须实现 lookups(request, model) get_filtered_query(query, value)
"""
def __init__(self, column: Any, *, title: str, parameter_name: str) -> None:
self.column = column
self.title = title
self.parameter_name = parameter_name
def lookups(self, request, model) -> list[tuple[str, str]]: # noqa: ARG002
return [
("all", "全部"),
("1h", "最近1小时"),
("24h", "最近24小时"),
("7d", "最近7天"),
("30d", "最近30天"),
]
def get_filtered_query(self, query, value): # type: ignore[no-untyped-def]
if not value or value == "all":
return query
now = datetime.utcnow()
delta: timedelta | None = None
if value == "1h":
delta = timedelta(hours=1)
elif value == "24h":
delta = timedelta(hours=24)
elif value == "7d":
delta = timedelta(days=7)
elif value == "30d":
delta = timedelta(days=30)
if delta is None:
return query
threshold = now - delta
# 兼容 SQLAlchemy Query / Select 两种风格
if hasattr(query, "where"):
return query.where(self.column >= threshold)
if hasattr(query, "filter"):
return query.filter(self.column >= threshold)
return query

View File

@ -8,11 +8,9 @@ from zoneinfo import ZoneInfo
from croniter import croniter
from markupsafe import Markup
from sqladmin import ModelView, action
from sqladmin.filters import BooleanFilter, OperationColumnFilter, StaticValuesFilter
from sqladmin.models import Request
from starlette.responses import RedirectResponse
from app.admin.filters import RecentDateTimeFilter
from app.db.models import Job, JobLog
from app.plugins.manager import load_job_class
from app.security.fernet import encrypt_json
@ -77,16 +75,6 @@ class JobAdmin(ModelView, model=Job):
# 列表页模板:加入每行 Run Now
list_template = "job_list.html"
# 搜索
column_searchable_list = [Job.id, Job.handler_path, Job.cron_expr]
# 筛选
column_filters = [
BooleanFilter(Job.enabled),
OperationColumnFilter(Job.handler_path),
RecentDateTimeFilter(Job.created_at, title="创建时间", parameter_name="created_at_recent"),
]
# 编辑页排除 secret_cfg避免回显密文由自定义模板额外渲染一个空输入框
# 注意SQLAdmin 这里需要字段名字符串(不是 SQLAlchemy Column 对象)
form_edit_rules = ["id", "enabled", "cron_expr", "handler_path", "public_cfg"]
@ -226,25 +214,6 @@ class JobLogAdmin(ModelView, model=JobLog):
# 为 JobLog 详情页单独指定模板(用于加入 Retry 按钮)
details_template = "joblog_details.html"
# 搜索
column_searchable_list = [JobLog.job_id, JobLog.message, JobLog.celery_task_id]
# 筛选
column_filters = [
StaticValuesFilter(
JobLog.status,
values=[
("RUNNING", "运行中"),
("SUCCESS", "成功"),
("FAILURE", "失败"),
("RETRY", "重试"),
],
title="状态",
),
OperationColumnFilter(JobLog.attempt),
RecentDateTimeFilter(JobLog.started_at, title="开始时间", parameter_name="started_at_recent"),
]
column_labels = {
"id": "日志ID",
"job_id": "任务ID",