diff --git a/backend/package/yuxi/repositories/scheduled/base.py b/backend/package/yuxi/repositories/scheduled/base.py index c19a10ee..1e949341 100644 --- a/backend/package/yuxi/repositories/scheduled/base.py +++ b/backend/package/yuxi/repositories/scheduled/base.py @@ -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) diff --git a/backend/package/yuxi/repositories/scheduled/scheduled_task_idempotency_repository.py b/backend/package/yuxi/repositories/scheduled/scheduled_task_idempotency_repository.py index c5a51aaf..384b82e2 100644 --- a/backend/package/yuxi/repositories/scheduled/scheduled_task_idempotency_repository.py +++ b/backend/package/yuxi/repositories/scheduled/scheduled_task_idempotency_repository.py @@ -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() diff --git a/backend/package/yuxi/repositories/scheduled/scheduled_task_repository.py b/backend/package/yuxi/repositories/scheduled/scheduled_task_repository.py index 25c4bacc..7e91167d 100644 --- a/backend/package/yuxi/repositories/scheduled/scheduled_task_repository.py +++ b/backend/package/yuxi/repositories/scheduled/scheduled_task_repository.py @@ -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`` 状态