58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
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
|
|
|