refactor(repo): 优化仓库层代码，新增多条件查询能力
1. 为count_deleted方法新增按deleted_at时间范围过滤的参数 2. 为幂等查询get方法新增operation参数以避免响应体错误重放 3. 优化定时任务更新方法的注释说明与排除字段逻辑 4. 新增活跃任务统计并更新handler聚合摘要查询 5. 为任务状态统计接口新增handler名称过滤能力
This commit is contained in:
parent
6ddc2a8c25
commit
202defb5f2
@ -155,9 +155,18 @@ class BaseRepository:
|
||||
result = await self.db.execute(stmt)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def count_deleted(self) -> int:
|
||||
"""统计软删除记录数。"""
|
||||
async def count_deleted(
|
||||
self,
|
||||
*,
|
||||
start_time: datetime | None = None,
|
||||
end_time: datetime | None = None,
|
||||
) -> int:
|
||||
"""统计软删除记录数，支持按 deleted_at 时间范围过滤。"""
|
||||
stmt = select(func.count()).select_from(self.model).where(self.model.is_deleted == 1)
|
||||
if start_time is not None:
|
||||
stmt = stmt.where(self.model.deleted_at >= start_time)
|
||||
if end_time is not None:
|
||||
stmt = stmt.where(self.model.deleted_at <= end_time)
|
||||
result = await self.db.execute(stmt)
|
||||
return int(result.scalar() or 0)
|
||||
|
||||
|
||||
@ -129,19 +129,24 @@ class ScheduledTaskIdempotencyRepository(BaseRepository):
|
||||
# 幂等查询
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def get(self, key: str) -> dict | None:
|
||||
async def get(self, key: str, *, operation: str | None = None) -> dict | None:
|
||||
"""查询幂等记录的响应体。
|
||||
|
||||
Args:
|
||||
key: 幂等键。
|
||||
operation: 操作类型（可选）。传入时额外校验记录中的 ``operation``
|
||||
字段与当前操作一致，避免不同 API 操作间因复用同一 key 导致
|
||||
响应体被错误重放（如 create 的响被 update 重放）。
|
||||
|
||||
Returns:
|
||||
``response_body`` 字段值（dict）；记录不存在或响应体未就绪
|
||||
（``response_body IS NULL``）时返回 ``None``。
|
||||
``response_body`` 字段值（dict）；记录不存在、响应体未就绪
|
||||
（``response_body IS NULL``）或 ``operation`` 不匹配时返回 ``None``。
|
||||
"""
|
||||
stmt = (
|
||||
select(self.model.response_body).where(self.model.idempotency_key == key).where(self.model.is_deleted == 0)
|
||||
)
|
||||
if operation is not None:
|
||||
stmt = stmt.where(self.model.operation == operation)
|
||||
result = await self.db.execute(stmt)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
@ -35,6 +35,7 @@ class HandlerSummary:
|
||||
|
||||
name: str
|
||||
task_count: int
|
||||
active_task_count: int
|
||||
last_active_at: datetime | None
|
||||
|
||||
|
||||
@ -226,25 +227,22 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
return tasks, total
|
||||
|
||||
async def update(self, task_id: str, data: dict[str, Any], *, commit: bool = True) -> ScheduledTask | None:
|
||||
"""按 ``task_id`` 更新任务配置字段（排除已软删除）。
|
||||
"""按 ``task_id`` 更新任务字段（排除已软删除）。
|
||||
|
||||
自动设置 ``updated_at=now``。从 ``data`` 中提取允许更新的字段，
|
||||
排除主键、审计字段与运行时状态字段。运行时状态字段
|
||||
（``consecutive_errors`` / ``last_run_at`` / ``next_run_at`` /
|
||||
``last_error`` / ``status``）由专用状态机方法管理，禁止通过本方法
|
||||
直接更新，避免绕过状态机约束：
|
||||
自动设置 ``updated_at=now``。从 ``data`` 中提取允许更新的字段，
|
||||
排除主键、审计字段与运行轨迹字段。运行轨迹字段
|
||||
（``consecutive_errors`` / ``last_run_at`` / ``last_error``）仍由
|
||||
``record_run_result`` 专用方法管理，禁止通过本方法直接更新；
|
||||
``status`` / ``next_run_at`` 由 ``update_task`` 用例在已校验状态机
|
||||
与调度语义后显式写入，避免 lower-level 调用方绕过约束。
|
||||
|
||||
- ``pause`` / ``resume`` —— active ↔ paused
|
||||
- ``mark_dead_letter`` —— → dead_letter
|
||||
- ``record_run_result`` —— 原子回写执行结果与 next_run_at
|
||||
Args:
|
||||
task_id: 业务任务标识。
|
||||
data: 待更新字段字典。
|
||||
commit: ``True`` 时提交事务，``False`` 时仅 flush。
|
||||
|
||||
Args:
|
||||
task_id: 业务任务标识。
|
||||
data: 待更新字段字典。
|
||||
commit: ``True`` 时提交事务，``False`` 时仅 flush。
|
||||
|
||||
Returns:
|
||||
更新后的 ``ScheduledTask`` 实例，或 None（任务不存在时）。
|
||||
Returns:
|
||||
更新后的 ``ScheduledTask`` 实例，或 None（任务不存在时）。
|
||||
"""
|
||||
excluded = {
|
||||
"task_id",
|
||||
@ -254,9 +252,7 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
"is_deleted",
|
||||
"consecutive_errors",
|
||||
"last_run_at",
|
||||
"next_run_at",
|
||||
"last_error",
|
||||
"status",
|
||||
}
|
||||
update_fields = {k: v for k, v in data.items() if k not in excluded}
|
||||
update_fields["updated_at"] = utc_now_naive()
|
||||
@ -372,11 +368,13 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
async def list_handler_summary(self) -> list[HandlerSummary]:
|
||||
"""返回 handler 维度的聚合摘要。
|
||||
|
||||
SQL 语义：``SELECT handler_name, COUNT(*) as task_count, MAX(last_run_at)
|
||||
as last_active_at FROM scheduled_tasks WHERE is_deleted=0 GROUP BY
|
||||
handler_name ORDER BY handler_name``。聚合查询用于 Admin API 展示
|
||||
各 handler 的任务数与最近执行时间，扫描全表但行数有限（handler 数 ×
|
||||
状态组合），性能可接受。
|
||||
SQL 语义：``SELECT handler_name,
|
||||
COUNT(*) as task_count,
|
||||
SUM(CASE WHEN status='active' THEN 1 ELSE 0 END) as active_task_count,
|
||||
MAX(last_run_at) as last_active_at
|
||||
FROM scheduled_tasks WHERE is_deleted=0 GROUP BY handler_name``。
|
||||
聚合查询用于 Admin API 展示各 handler 的任务数、活跃任务数与最近执行时间，
|
||||
扫描全表但行数有限（handler 数 × 状态组合），性能可接受。
|
||||
|
||||
``last_active_at`` 取 ``last_run_at``（最近一次执行开始时间）而非
|
||||
``updated_at``，避免配置变更等非执行操作刷新活跃时间导致语义失真。
|
||||
@ -388,6 +386,7 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
select(
|
||||
ScheduledTask.handler_name,
|
||||
func.count().label("task_count"),
|
||||
func.sum(func.case((ScheduledTask.status == "active", 1), else_=0)).label("active_task_count"),
|
||||
func.max(ScheduledTask.last_run_at).label("last_active_at"),
|
||||
)
|
||||
.where(self._not_deleted())
|
||||
@ -399,6 +398,7 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
HandlerSummary(
|
||||
name=row.handler_name,
|
||||
task_count=int(row.task_count),
|
||||
active_task_count=int(row.active_task_count or 0),
|
||||
last_active_at=row.last_active_at,
|
||||
)
|
||||
for row in result.all()
|
||||
@ -409,18 +409,20 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
*,
|
||||
owner_scope: str | None = None,
|
||||
owner_id: str | None = None,
|
||||
handler_name: str | None = None,
|
||||
) -> dict[str, int]:
|
||||
"""按状态统计任务数，返回 ``{active: N, paused: N, dead_letter: N}``。
|
||||
|
||||
用于 Admin API 仪表盘展示任务状态分布。SQL 语义：
|
||||
``SELECT status, COUNT(*) FROM scheduled_tasks WHERE is_deleted=0
|
||||
[AND owner_scope=? AND owner_id=?] GROUP BY status``，一次查询
|
||||
获取所有状态分布，避免上层多次调用 ``list(status=..., page_size=1)``
|
||||
[AND owner_scope=? AND owner_id=? AND handler_name=?] GROUP BY status``，
|
||||
一次查询获取所有状态分布，避免上层多次调用 ``list(status=..., page_size=1)``
|
||||
获取 total 的低效方案。
|
||||
|
||||
Args:
|
||||
owner_scope: 可选归属范围过滤（system / business）。
|
||||
owner_id: 可选业务方 ID 过滤。
|
||||
handler_name: 可选 handler 名称过滤。
|
||||
|
||||
Returns:
|
||||
状态 → 任务数的映射；未出现的状态不在字典中（调用方按需 ``.get`` 补 0）。
|
||||
@ -434,6 +436,8 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
stmt = stmt.where(ScheduledTask.owner_scope == owner_scope)
|
||||
if owner_id is not None:
|
||||
stmt = stmt.where(ScheduledTask.owner_id == owner_id)
|
||||
if handler_name is not None:
|
||||
stmt = stmt.where(ScheduledTask.handler_name == handler_name)
|
||||
result = await self.db.execute(stmt)
|
||||
return {row.status: int(row.count) for row in result.all()}
|
||||
|
||||
@ -502,9 +506,7 @@ class ScheduledTaskRepository(BaseRepository):
|
||||
# 状态机转换
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def pause(
|
||||
self, task_id: str, *, updated_by: str | None = None, commit: bool = True
|
||||
) -> ScheduledTask | None:
|
||||
async def pause(self, task_id: str, *, updated_by: str | None = None, commit: bool = True) -> ScheduledTask | None:
|
||||
"""暂停任务（``active`` → ``paused``）。
|
||||
|
||||
仅 ``active`` 状态的任务可暂停，``paused`` / ``dead_letter`` 状态
|
||||
|
||||
Loading…
Reference in New Issue
Block a user