定时任务接口:增删改查、预览、变量清单、立即运行、触发历史和测试 Webhook

- 按前后端约定的 Schedule / Fire 形状返回;Webhook 地址和签名密钥加密存放,接口只给主机和末 4 位、有没有密钥
- 创建、修改时校验名称、cron、时区、工作流存在、正式运行要有发布版本(没有就说明原因、建议改探索运行)、输入
  模板和入口字段;不对回 422 {detail(中文), field},请求体自己解析,连 JSON 写坏了也回中文
- PATCH 带 version,条件 UPDATE 做乐观锁,对不上回 409;webhook.url 省略表示不改、空串或 null 表示删除,secret
  省略不改、空串去掉;只停用、改名不重新校验工作流——工作流撤了发布也要能先把任务停下来
- 预览给下几次触发(UTC)和按下一次触发渲染的输入,带工作流时只查字段名、不查缺的必填项(用户还在填)
- 立即运行和到点触发同一套规则(重叠跳过、正式运行只从当前发布版本发起),运行记在点按钮的人名下;触发时间
  取到秒,同一秒连点交回同一条,不会发起两次
- 触发历史按创建时间倒序翻页,Fire 带运行现在的状态和推送记录;删除任务连同触发历史和推送记录,发起过的运行留着
This commit is contained in:
ThearchyHelios
2026-10-05 08:43:44 +08:00
parent a2a19df9df
commit 6d11145a39
2 changed files with 981 additions and 3 deletions
+517 -3
View File
@@ -1,9 +1,523 @@
"""定时任务:按周期自动发起工作流运行。
"""定时任务:按周期自动发起工作流运行,结束、失败、停在审批时推送 Webhook。
目前只登记了路由前缀,接口由定时任务的工作包补上;后台的调度循环在 app/scheduler 里。
形状按前后端约定(Schedule、Fire 两种对象,见 _schedule_out、_fire_outs)。Webhook 地址和签名密钥加密存放,接口
只给主机和末 4 位(url_hint)、有没有密钥(has_secret)。请求体自己解析、自己校验:错误一律回中文 detail,表单
据 field 把报错落到那一格(422 {detail, code, field},同 api/coded.py)。后台的调度循环在 app/scheduler 里。
"""
from __future__ import annotations
from fastapi import APIRouter
from datetime import datetime
from typing import Any
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response
from sqlalchemy import and_, delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.coded import CodedHTTPException
from app.api.runs import actor_of
from app.core.crypto import decrypt, encrypt
from app.db.base import get_session, utcnow
from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion
from app.scheduler import cron, service, webhook
router = APIRouter(prefix="/api/schedules", tags=["schedules"])
#: 请求不合规(422)的机器码。field 是出错的那一项:name、workflow_id、cron、timezone、run_class、input_template、
#: enabled、overlap、misfire_grace_s、webhook、version、limit、before
SCHEDULE_INVALID = "schedule_invalid"
_GONE = "定时任务不存在,可能已被删除"
#: 错过的触发在多少秒内补发:默认 1 小时,最长 7 天
DEFAULT_GRACE_S = 3600
MAX_GRACE_S = 7 * 24 * 3600
#: 预览最多给几次
MAX_PREVIEW = 20
def _invalid(detail: str, field: str | None = None) -> CodedHTTPException:
return CodedHTTPException(422, detail, SCHEDULE_INVALID, field=field)
def _iso(moment: datetime | None) -> str | None:
return service.as_utc(moment).isoformat() if moment is not None else None
async def _body(request: Request) -> dict[str, Any]:
try:
data = await request.json()
except ValueError:
data = None
if not isinstance(data, dict):
raise _invalid("请求内容需要是 JSON 对象")
return data
# --------------------------------------------------------------------------
# 逐项校验:返回规范化之后的值,不对抛 422(带 field)
# --------------------------------------------------------------------------
def _name(value: Any) -> str:
text = value.strip() if isinstance(value, str) else ""
if not text:
raise _invalid("请填写定时任务的名称", "name")
if len(text) > 200:
raise _invalid("定时任务的名称不能超过 200 个字符", "name")
return text
def _workflow_id(value: Any) -> str:
if not isinstance(value, str) or not value.strip():
raise _invalid("请选择要定时运行的工作流", "workflow_id")
return value.strip()
def _cron(value: Any) -> str:
try:
return cron.parse(value if isinstance(value, str) else "").expr
except cron.CronError as e:
raise _invalid(str(e), "cron") from e
def _timezone(value: Any) -> str:
try:
return cron.timezone_name(value if isinstance(value, str) else "")
except cron.TimezoneError as e:
raise _invalid(str(e), "timezone") from e
def _run_class(value: Any) -> str:
if value not in service.RUN_CLASSES:
raise _invalid(f"运行类别需要是 formal(正式运行)或 exploratory(探索运行),当前为「{value}」", "run_class")
return value
def _template(value: Any) -> dict[str, str]:
try:
return service.check_template(value)
except service.TemplateError as e:
raise _invalid(str(e), "input_template") from e
def _enabled(value: Any) -> bool:
if not isinstance(value, bool):
raise _invalid(f"启用状态需要是 true 或 false,当前为「{value}」", "enabled")
return value
def _overlap(value: Any) -> str:
if value != "skip":
raise _invalid(f"重叠处理目前只支持 skip(上一次运行还没结束时跳过这次触发),当前为「{value}」", "overlap")
return value
def _grace(value: Any) -> int:
if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= MAX_GRACE_S:
raise _invalid(f"补发宽限需要是 0 到 {MAX_GRACE_S} 之间的整数秒数(最长 7 天),当前为「{value}」",
"misfire_grace_s")
return value
def _format(value: Any) -> str:
if value not in webhook.FORMATS:
raise _invalid(f"不支持的 Webhook 格式「{value}」,可选的格式:{'、'.join(webhook.FORMATS)}", "webhook")
return value
def _events(value: Any) -> list[str]:
if not isinstance(value, list) or any(e not in webhook.EVENTS for e in value):
raise _invalid("Webhook 的通知事件只能从 finished(运行完成)、failed(运行失败)、awaiting_approval(等待审批)"
f"中选择,当前为「{value}」", "webhook")
return [e for e in webhook.EVENTS if e in value]
def _url(value: Any) -> str:
try:
return webhook.check_url(value)
except ValueError as e:
raise _invalid(str(e), "webhook") from e
def _secret(value: Any) -> str | None:
if value is None or (isinstance(value, str) and not value.strip()):
return None
if not isinstance(value, str):
raise _invalid("Webhook 签名密钥需要是文本", "webhook")
return value.strip()
def _blank(value: Any) -> bool:
return value is None or (isinstance(value, str) and not value.strip())
def _new_webhook(value: Any) -> dict[str, Any] | None:
"""创建时的 webhook:null 或 url 为空表示不配;否则校验格式、地址、密钥、事件(事件省略表示三种都要)。"""
if value is None:
return None
if not isinstance(value, dict):
raise _invalid("Webhook 设置需要是一个对象,或者为空表示不推送", "webhook")
if _blank(value.get("url")):
return None
return {"format": _format(value.get("format", "generic")), "url": _url(value.get("url")),
"secret": _secret(value.get("secret")), "events": _events(value.get("events", list(webhook.EVENTS)))}
def _webhook_columns(hook: dict[str, Any] | None) -> dict[str, Any]:
if hook is None:
return {"webhook_format": None, "webhook_url": None, "webhook_secret": None, "webhook_events": []}
return {"webhook_format": hook["format"], "webhook_url": encrypt(hook["url"]),
"webhook_secret": encrypt(hook["secret"]) if hook["secret"] else None, "webhook_events": hook["events"]}
def _patched_webhook(row: Schedule, value: Any) -> dict[str, Any]:
"""PATCH 的 webhook → 要写的列。null 或 url 写空串:删除整个 Webhook;url 省略:地址不变;secret 省略:密钥不变、
写空串:去掉密钥。"""
if value is None:
return _webhook_columns(None)
if not isinstance(value, dict):
raise _invalid("Webhook 设置需要是一个对象,或者为空表示不推送", "webhook")
if "url" in value and _blank(value.get("url")):
return _webhook_columns(None)
has_url = bool(row.webhook_url)
if "url" not in value and not has_url:
raise _invalid("请填写 Webhook 地址", "webhook")
out: dict[str, Any] = {
"webhook_format": _format(value.get("format", row.webhook_format if has_url else "generic")),
"webhook_events": _events(value.get("events", list(row.webhook_events or []) if has_url
else list(webhook.EVENTS))),
}
if "url" in value:
out["webhook_url"] = encrypt(_url(value.get("url")))
if "secret" in value:
secret = _secret(value.get("secret"))
out["webhook_secret"] = encrypt(secret) if secret else None
return out
async def _check_workflow(session: AsyncSession, workflow_id: str, run_class: str, template: dict[str, str]) -> Workflow:
"""工作流要在;正式运行要有发布版本;模板里的字段名对得上入口字段,必填项都有(按这次要用的那张图查)。"""
workflow = await session.get(Workflow, workflow_id)
if workflow is None:
raise _invalid("工作流不存在,可能已被删除", "workflow_id")
graph: Any = workflow.graph
if run_class == "formal":
if not workflow.published_version:
raise _invalid(f"「{workflow.name}」还没有发布版本,正式运行只能从发布版本发起。"
"请先在编排页发布,或改为探索运行", "run_class")
graph = await _published_graph(session, workflow) or graph
try:
service.check_fields(template, service.input_fields(graph), workflow.name)
except service.TemplateError as e:
raise _invalid(str(e), "input_template") from e
return workflow
async def _published_graph(session: AsyncSession, workflow: Workflow) -> Any:
return await session.scalar(select(WorkflowVersion.graph).where(
WorkflowVersion.workflow_id == workflow.id, WorkflowVersion.version == workflow.published_version))
# --------------------------------------------------------------------------
# 对外的形状
# --------------------------------------------------------------------------
def _description(row: Schedule) -> str:
try:
return cron.describe(row.cron, row.timezone)
except cron.CronError:
return f"按 cron 表达式 {row.cron}({row.timezone})"
def _webhook_out(row: Schedule) -> dict[str, Any] | None:
if not row.webhook_url:
return None
plain = decrypt(row.webhook_url)
return {
"format": row.webhook_format,
# 解不开(加密密钥换过)时不猜:照实说,请人重新填
"url_hint": webhook.url_hint(plain) if plain else "(地址无法解密,请重新填写)",
"has_secret": bool(row.webhook_secret),
"events": [e for e in webhook.EVENTS if e in (row.webhook_events or [])],
}
async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[dict[str, Any]]:
"""Fire 对象:run_status 是运行现在的状态,notified 是这次触发的推送记录 [{event, status, at}](按时间先后)。"""
run_ids = [f.run_id for f in fires if f.run_id]
statuses = dict((await session.execute(select(Run.id, Run.status).where(Run.id.in_(run_ids)))).all()) \
if run_ids else {}
notified: dict[str, list[dict[str, Any]]] = {}
if fires:
rows = (await session.execute(
select(WebhookDelivery).where(WebhookDelivery.fire_id.in_([f.id for f in fires]))
.order_by(WebhookDelivery.created_at)
)).scalars()
for d in rows:
notified.setdefault(d.fire_id, []).append({"event": d.event, "status": d.status, "at": _iso(d.created_at)})
return [{
"id": f.id,
"fire_at": _iso(f.fire_at),
"kind": f.kind,
"status": f.status,
"run_id": f.run_id,
"run_status": statuses.get(f.run_id) if f.run_id else None,
"reason": f.reason,
"notified": notified.get(f.id, []),
"created_at": _iso(f.created_at),
} for f in fires]
async def _schedule_outs(session: AsyncSession, rows: list[Schedule]) -> list[dict[str, Any]]:
ids = [r.id for r in rows]
names: dict[str, str] = {}
last: dict[str, dict[str, Any]] = {}
if rows:
names = dict((await session.execute(
select(Workflow.id, Workflow.name).where(Workflow.id.in_({r.workflow_id for r in rows})))).all())
newest = (select(ScheduleFire.schedule_id, func.max(ScheduleFire.created_at).label("at"))
.where(ScheduleFire.schedule_id.in_(ids)).group_by(ScheduleFire.schedule_id).subquery())
fires = list((await session.execute(select(ScheduleFire).join(newest, and_(
ScheduleFire.schedule_id == newest.c.schedule_id, ScheduleFire.created_at == newest.c.at)))).scalars())
for out, fire in zip(await _fire_outs(session, fires), fires):
last.setdefault(fire.schedule_id, out)
return [{
"id": r.id,
"name": r.name,
"workflow_id": r.workflow_id,
# 工作流还在就给它现在的名字;删掉之后用保存时记下的那一份
"workflow_name": names.get(r.workflow_id) or r.workflow_name,
"cron": r.cron,
"timezone": r.timezone,
"run_class": r.run_class,
"input_template": dict(r.input_template or {}),
"enabled": r.enabled,
"overlap": r.overlap,
"misfire_grace_s": r.misfire_grace_s,
"webhook": _webhook_out(r),
"next_run_at": _iso(r.next_run_at),
"last_fire": last.get(r.id),
"description_text": _description(r),
"created_by": r.created_by,
"created_at": _iso(r.created_at),
"updated_at": _iso(r.updated_at),
"version": r.version,
} for r in rows]
async def _schedule_out(session: AsyncSession, row: Schedule) -> dict[str, Any]:
return (await _schedule_outs(session, [row]))[0]
async def _get(session: AsyncSession, schedule_id: str) -> Schedule:
row = await session.get(Schedule, schedule_id)
if row is None:
raise HTTPException(404, _GONE)
return row
# --------------------------------------------------------------------------
# 固定路径(要排在 /{schedule_id} 前面)
# --------------------------------------------------------------------------
@router.get("/variables")
async def list_variables(tz: str | None = Query(default=None, alias="timezone")) -> list[dict[str, str]]:
"""输入模板能用的变量:[{name, label, example}]。示例按现在这一刻、timezone(默认 Asia/Shanghai)触发算。"""
zone = _timezone(tz) if tz is not None else service.DEFAULT_TIMEZONE
return service.variables(tz=zone)
@router.post("/preview")
async def preview(request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""{cron, timezone, input_template?, workflow_id?, n?=5} → {description_text, next: [ISO(UTC)…], inputs}。
inputs 是按下一次触发渲染的输入。带了 workflow_id 时对照入口字段的名字;必填项没填不在这里报——用户还在填。
"""
body = await _body(request)
cron_expr, tz = _cron(body.get("cron")), _timezone(body.get("timezone"))
template = _template(body.get("input_template"))
n = body.get("n", 5)
n = min(max(n, 1), MAX_PREVIEW) if isinstance(n, int) and not isinstance(n, bool) else 5
workflow_id = body.get("workflow_id")
if isinstance(workflow_id, str) and workflow_id.strip():
workflow = await session.get(Workflow, workflow_id.strip())
if workflow is not None:
# 还不知道按哪一类运行:发布版本和画布的字段都认
published = await _published_graph(session, workflow) if workflow.published_version else None
fields = service.input_fields(workflow.graph) + service.input_fields(published)
try:
service.check_fields(template, fields, workflow.name, require=False)
except service.TemplateError as e:
raise _invalid(str(e), "input_template") from e
upcoming = cron.preview(cron_expr, tz, n)
first = upcoming[0] if upcoming else utcnow()
return {
"description_text": cron.describe(cron_expr, tz),
"next": [_iso(m) for m in upcoming],
"inputs": service.render_inputs(template, service.template_values(first, tz, cron_expr)),
}
# --------------------------------------------------------------------------
# 增删改查
# --------------------------------------------------------------------------
@router.get("")
async def list_schedules(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
rows = list((await session.execute(select(Schedule).order_by(Schedule.created_at.desc()))).scalars())
return {"items": await _schedule_outs(session, rows)}
@router.post("", status_code=201)
async def create_schedule(request: Request, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
body = await _body(request)
name = _name(body.get("name"))
workflow_id = _workflow_id(body.get("workflow_id"))
cron_expr, tz = _cron(body.get("cron")), _timezone(body.get("timezone"))
run_class = _run_class(body.get("run_class", "formal"))
template = _template(body.get("input_template"))
enabled = _enabled(body.get("enabled", True))
overlap = _overlap(body.get("overlap", "skip"))
grace = _grace(body.get("misfire_grace_s", DEFAULT_GRACE_S))
hook = _new_webhook(body.get("webhook"))
workflow = await _check_workflow(session, workflow_id, run_class, template)
row = Schedule(name=name, workflow_id=workflow_id, workflow_name=workflow.name, cron=cron_expr, timezone=tz,
run_class=run_class, input_template=template, enabled=enabled, overlap=overlap,
misfire_grace_s=grace, next_run_at=service.next_run(cron_expr, tz, enabled),
created_by=actor_of(x_actor), **_webhook_columns(hook))
session.add(row)
await session.commit()
await session.refresh(row)
return await _schedule_out(session, row)
@router.get("/{schedule_id}")
async def get_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
return await _schedule_out(session, await _get(session, schedule_id))
@router.patch("/{schedule_id}")
async def update_schedule(schedule_id: str, request: Request,
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""改任意可改的项,带上读到的 version;对不上回 409。只读的项(id、next_run_at、last_fire……)带了也不理。
改了 cron、时区或启用状态就重算下一次触发(从现在算,不补停用期间错过的)。改了工作流、运行类别或输入模板
就按改后的组合重新校验;只停用、改名不查工作流——工作流后来撤了发布,也要能先把任务停下来。
"""
body = await _body(request)
row = await _get(session, schedule_id)
version = body.get("version")
if isinstance(version, bool) or not isinstance(version, int):
raise _invalid("保存时需要带上读到的版本号 version,请刷新后重试", "version")
if version != row.version:
raise HTTPException(409, f"这个定时任务已被修改(当前是第 {row.version} 版,提交的是第 {version} 版),"
"请刷新后再保存")
values: dict[str, Any] = {}
if "name" in body:
values["name"] = _name(body["name"])
if "workflow_id" in body:
values["workflow_id"] = _workflow_id(body["workflow_id"])
if "cron" in body:
values["cron"] = _cron(body["cron"])
if "timezone" in body:
values["timezone"] = _timezone(body["timezone"])
if "run_class" in body:
values["run_class"] = _run_class(body["run_class"])
if "input_template" in body:
values["input_template"] = _template(body["input_template"])
if "enabled" in body:
values["enabled"] = _enabled(body["enabled"])
if "overlap" in body:
values["overlap"] = _overlap(body["overlap"])
if "misfire_grace_s" in body:
values["misfire_grace_s"] = _grace(body["misfire_grace_s"])
if "webhook" in body:
values |= _patched_webhook(row, body["webhook"])
def merged(key: str) -> Any:
return values.get(key, getattr(row, key))
if {"workflow_id", "run_class", "input_template"} & values.keys():
workflow = await _check_workflow(session, merged("workflow_id"), merged("run_class"), merged("input_template"))
values["workflow_name"] = workflow.name
if {"cron", "timezone", "enabled"} & values.keys():
values["next_run_at"] = service.next_run(merged("cron"), merged("timezone"), merged("enabled"))
# 条件 UPDATE 才是真的乐观锁:两个人同时读到第 3 版、同时保存,只有先到的那个改得动
result = await session.execute(
update(Schedule).where(Schedule.id == schedule_id, Schedule.version == version)
.values(**values, version=version + 1).execution_options(synchronize_session=False))
if result.rowcount != 1:
await session.rollback()
raise HTTPException(409, "这个定时任务刚被别人修改过,请刷新后再保存")
await session.commit()
session.expire_all()
return await _schedule_out(session, await _get(session, schedule_id))
@router.delete("/{schedule_id}", status_code=204)
async def delete_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> Response:
"""删除任务,连同它的触发历史和推送记录。它发起过的运行留着(运行上的来源 id 也留着):那是追溯凭证。"""
await _get(session, schedule_id)
# 外键是 ON DELETE CASCADE;这里照样按顺序删一遍,不依赖连接上开没开外键约束
await session.execute(delete(WebhookDelivery).where(WebhookDelivery.schedule_id == schedule_id))
await session.execute(delete(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id))
await session.execute(delete(Schedule).where(Schedule.id == schedule_id))
await session.commit()
return Response(status_code=204)
# --------------------------------------------------------------------------
# 立即运行、触发历史、测试 Webhook
# --------------------------------------------------------------------------
@router.post("/{schedule_id}/run")
async def run_now(schedule_id: str, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
"""立即触发一次(kind=manual),停用的任务也可以。规则和到点触发相同:上一次运行没结束就跳过,正式运行只从当前
发布版本发起。运行记在点按钮的人名下,没有署名时记在「定时任务:<名字>」名下。返回 {fire, run_id}。
触发时间取到秒:连点两下落在同一秒的,交回同一条触发记录,不会发起两次。"""
await _get(session, schedule_id)
fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual", actor=actor_of(x_actor))
if fire is None:
raise HTTPException(404, _GONE)
return {"fire": (await _fire_outs(session, [fire]))[0], "run_id": fire.run_id}
@router.get("/{schedule_id}/fires")
async def list_fires(schedule_id: str, limit: str | None = None, before: str | None = None,
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""触发历史,新的在前:{items: [Fire], next_before}。翻页把 next_before 原样作为 before 传回来;没有更多时为空。"""
await _get(session, schedule_id)
size = 20
if limit is not None:
if not limit.strip().isdigit() or not 1 <= int(limit) <= 100:
raise _invalid(f"每页条数需要是 1 到 100 之间的整数,当前为「{limit}」", "limit")
size = int(limit)
stmt = select(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id)
if before is not None:
try:
cursor = datetime.fromisoformat(before.strip().replace("Z", "+00:00"))
except ValueError as e:
raise _invalid(f"翻页位置需要是接口返回的 next_before,当前为「{before}」", "before") from e
stmt = stmt.where(ScheduleFire.created_at < service.as_utc(cursor))
rows = list((await session.execute(
stmt.order_by(ScheduleFire.created_at.desc(), ScheduleFire.id.desc()).limit(size + 1))).scalars())
items = rows[:size]
return {"items": await _fire_outs(session, items),
"next_before": _iso(items[-1].created_at) if len(rows) > size else None}
@router.post("/{schedule_id}/webhook/test")
async def test_webhook(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""按保存的 Webhook 发一条测试消息(不重试):{ok, http_status, message}。失败的说明里只有主机名。"""
await _get(session, schedule_id)
try:
return await webhook.send_test(schedule_id)
except KeyError as e:
raise HTTPException(404, _GONE) from e
except ValueError as e:
raise _invalid(str(e), "webhook") from e
+464
View File
@@ -0,0 +1,464 @@
"""定时任务接口(api/schedules.py),按前后端约定的形状:
- Schedule:{id, name, workflow_id, workflow_name, cron, timezone, run_class, input_template, enabled, overlap,
misfire_grace_s, webhook: {format, url_hint, has_secret, events} | null, next_run_at, last_fire, description_text,
created_by, created_at, updated_at, version};Webhook 地址和密钥加密存放,接口只给主机和末 4 位
- Fire:{id, fire_at, kind, status, run_id, run_status, reason, notified: [{event, status, at}], created_at}
- 创建、修改时校验 cron、时区、工作流存在、正式运行要有发布版本、输入模板;不对回 422 {detail(中文), field}
- PATCH 带 version,对不上回 409;webhook.url 省略表示不改,空串表示删除 Webhook
- 预览下几次触发和渲染后的输入;变量清单;立即运行(同一套重叠规则);触发历史翻页;测试 Webhook
工作流、任务名、地址、密钥都是虚构的。
"""
from __future__ import annotations
import asyncio
import json
import uuid
from datetime import datetime, timedelta, timezone
from urllib.parse import quote
from zoneinfo import ZoneInfo
import httpx
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, func, select
from app.core.crypto import decrypt
from app.db.base import SessionLocal
from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion
from app.api import schedules as schedules_api
from app.engine.runner import run_manager
from app.main import app
from app.scheduler import webhook
UTC = timezone.utc
HOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/7c6b5a49-dddd-eeee-ffff-1234567890ab"
SECRET = "SECdemo-api-not-real-7"
SCHEDULE_KEYS = {"id", "name", "workflow_id", "workflow_name", "cron", "timezone", "run_class", "input_template",
"enabled", "overlap", "misfire_grace_s", "webhook", "next_run_at", "last_fire", "description_text",
"created_by", "created_at", "updated_at", "version"}
FIRE_KEYS = {"id", "fire_at", "kind", "status", "run_id", "run_status", "reason", "notified", "created_at"}
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
async with SessionLocal() as session:
await session.execute(delete(Schedule))
await session.commit()
yield
await run_manager.shutdown()
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
def node(nid, ntype, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
def graph(*fields: dict) -> dict:
return {"nodes": [node("start", "input", fields=list(fields)),
node("out", "output", fields=[{"name": "r", "value": "完成 {{ input.stat_date }}"}])],
"edges": [{"source": "start", "target": "out"}]}
DATE_FIELD = {"name": "stat_date", "required": True}
async def workflow(*, published: bool = True, fields=(DATE_FIELD,)) -> tuple[str, str]:
name = f"门店周报-{uuid.uuid4().hex[:6]}"
async with SessionLocal() as session:
wf = Workflow(name=name, graph=graph(*fields), status="published" if published else "draft",
published_version=1 if published else None)
session.add(wf)
await session.flush()
session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph(*fields)))
await session.commit()
return wf.id, name
def payload(wf_id: str, **over) -> dict:
base = {"name": "周一早报", "workflow_id": wf_id, "cron": "0 9 * * 1", "timezone": "Asia/Shanghai",
"run_class": "formal", "input_template": {"stat_date": "{{ schedule.yesterday }}"}, "enabled": True,
"overlap": "skip", "misfire_grace_s": 3600,
"webhook": {"format": "feishu", "url": HOOK, "secret": SECRET, "events": ["finished", "failed"]}}
base.update(over)
return base
async def create(client, wf_id: str, **over) -> dict:
res = await client.post("/api/schedules", json=payload(wf_id, **over),
headers={"X-Actor": quote("李小明")})
assert res.status_code == 201, res.text
return res.json()
async def settle(run_id: str) -> None:
for _ in range(400):
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None:
await run_manager.wait_idle(run_id)
return
await asyncio.sleep(0.03)
raise AssertionError(f"运行 {run_id} 停在 {row.status}")
# --------------------------------------------------------------------------
# 增删改查
# --------------------------------------------------------------------------
async def test_create_returns_the_agreed_shape_and_hides_the_address(client):
wf, wf_name = await workflow()
res = await client.post("/api/schedules", json=payload(wf), headers={"X-Actor": quote("李小明")})
assert res.status_code == 201
body = res.json()
assert set(body) == SCHEDULE_KEYS
assert body["workflow_name"] == wf_name and body["created_by"] == "李小明" and body["version"] == 1
assert body["webhook"] == {"format": "feishu", "url_hint": "open.feishu.cn/…90ab", "has_secret": True,
"events": ["finished", "failed"]}
assert body["description_text"] == "每周一 09:00(Asia/Shanghai)"
assert body["last_fire"] is None and body["overlap"] == "skip" and body["misfire_grace_s"] == 3600
next_at = datetime.fromisoformat(body["next_run_at"])
assert next_at.utcoffset().total_seconds() == 0 and next_at.astimezone(UTC).hour == 1
assert next_at > datetime.now(UTC)
# 地址、密钥一个字都不回到浏览器
assert "7c6b5a49" not in res.text and SECRET not in res.text and "/hook/" not in res.text
async with SessionLocal() as session:
row = await session.get(Schedule, body["id"])
assert row.webhook_url.startswith("enc::") and decrypt(row.webhook_url) == HOOK
assert row.webhook_secret.startswith("enc::") and decrypt(row.webhook_secret) == SECRET
async def test_list_and_get(client):
wf, _ = await workflow()
made = await create(client, wf)
listed = (await client.get("/api/schedules")).json()
assert set(listed) == {"items"} and [s["id"] for s in listed["items"]] == [made["id"]]
got = await client.get(f"/api/schedules/{made['id']}")
assert got.status_code == 200 and got.json() == made
async def test_patch_updates_fields_and_bumps_the_version(client):
wf, _ = await workflow()
made = await create(client, wf)
res = await client.patch(f"/api/schedules/{made['id']}",
json={"name": "每日早报", "cron": "30 8 * * *", "version": 1})
assert res.status_code == 200, res.text
body = res.json()
assert (body["name"], body["cron"], body["version"]) == ("每日早报", "30 8 * * *", 2)
assert body["description_text"] == "每天 08:30(Asia/Shanghai)"
assert datetime.fromisoformat(body["next_run_at"]).astimezone(UTC).minute == 30
# 停用:没有下一次
paused = (await client.patch(f"/api/schedules/{made['id']}", json={"enabled": False, "version": 2})).json()
assert paused["enabled"] is False and paused["next_run_at"] is None and paused["version"] == 3
resumed = (await client.patch(f"/api/schedules/{made['id']}", json={"enabled": True, "version": 3})).json()
assert resumed["next_run_at"] is not None
async def test_stale_version_is_a_conflict(client):
wf, _ = await workflow()
made = await create(client, wf)
assert (await client.patch(f"/api/schedules/{made['id']}", json={"name": "甲", "version": 1})).status_code == 200
res = await client.patch(f"/api/schedules/{made['id']}", json={"name": "乙", "version": 1})
assert res.status_code == 409 and "刷新" in res.json()["detail"]
missing = await client.patch(f"/api/schedules/{made['id']}", json={"name": "丙"})
assert missing.status_code == 422 and missing.json()["field"] == "version"
assert (await client.get(f"/api/schedules/{made['id']}")).json()["name"] == "甲"
async def test_patch_webhook_keeps_or_removes_the_address(client):
wf, _ = await workflow()
made = await create(client, wf)
sid = made["id"]
# 不带 url:地址不变,只改事件和格式
res = await client.patch(f"/api/schedules/{sid}", json={
"webhook": {"format": "dingtalk", "events": ["awaiting_approval"]}, "version": 1})
assert res.status_code == 200, res.text
assert res.json()["webhook"] == {"format": "dingtalk", "url_hint": "open.feishu.cn/…90ab", "has_secret": True,
"events": ["awaiting_approval"]}
async with SessionLocal() as session:
assert decrypt((await session.get(Schedule, sid)).webhook_url) == HOOK
# 密钥写空串:去掉签名
res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"secret": ""}, "version": 2})
assert res.json()["webhook"]["has_secret"] is False
# url 写空串:删除整个 Webhook
res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"url": ""}, "version": 3})
assert res.json()["webhook"] is None
async with SessionLocal() as session:
row = await session.get(Schedule, sid)
assert (row.webhook_url, row.webhook_secret, row.webhook_format) == (None, None, None)
# 没有地址时只给格式不行
res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"format": "wecom"}, "version": 4})
assert res.status_code == 422 and res.json()["field"] == "webhook"
# 重新配上
res = await client.patch(f"/api/schedules/{sid}", json={
"webhook": {"format": "wecom", "url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=abc-0000-9f9f"},
"version": 4})
assert res.json()["webhook"] == {"format": "wecom", "url_hint": "qyapi.weixin.qq.com/…9f9f", "has_secret": False,
"events": ["finished", "failed", "awaiting_approval"]}
async def test_webhook_null_removes_it(client):
wf, _ = await workflow()
made = await create(client, wf)
res = await client.patch(f"/api/schedules/{made['id']}", json={"webhook": None, "version": 1})
assert res.json()["webhook"] is None
async def test_delete(client):
wf, _ = await workflow()
made = await create(client, wf)
sid = made["id"]
ran = (await client.post(f"/api/schedules/{sid}/run")).json()
await settle(ran["run_id"])
res = await client.delete(f"/api/schedules/{sid}")
assert res.status_code == 204
gone = await client.get(f"/api/schedules/{sid}")
assert gone.status_code == 404 and gone.json()["detail"] == "定时任务不存在,可能已被删除"
async with SessionLocal() as session:
assert await session.scalar(select(func.count()).select_from(ScheduleFire)
.where(ScheduleFire.schedule_id == sid)) == 0
# 运行记录留着,来源 id 也留着:它是追溯凭证
assert (await session.get(Run, ran["run_id"])).schedule_id == sid
@pytest.mark.parametrize("method, path", [
("get", ""), ("patch", ""), ("delete", ""), ("post", "/run"), ("get", "/fires"), ("post", "/webhook/test")])
async def test_unknown_schedule_is_404_in_chinese(client, method, path):
kwargs = {"json": {"version": 1}} if method == "patch" else {}
res = await getattr(client, method)(f"/api/schedules/nope{uuid.uuid4().hex[:6]}{path}", **kwargs)
assert res.status_code == 404 and res.json()["detail"] == "定时任务不存在,可能已被删除"
# --------------------------------------------------------------------------
# 校验:422 {detail, field}
# --------------------------------------------------------------------------
@pytest.mark.parametrize("over, field, fragment", [
({"name": " "}, "name", "名称"),
({"cron": "0 25 * * *"}, "cron", "「时」"),
({"cron": "0 0 30 2 *"}, "cron", "永远不会触发"),
({"timezone": "Mars/Olympus"}, "timezone", "时区"),
({"run_class": "weekly"}, "run_class", "正式运行"),
({"input_template": {"stat_date": "{{ schedule.yesterdy }}"}}, "input_template", "schedule.yesterdy"),
({"input_template": {"stat_dat": "{{ schedule.yesterday }}"}}, "input_template", "stat_dat"),
({"input_template": {}}, "input_template", "必填"),
({"misfire_grace_s": -1}, "misfire_grace_s", "宽限"),
({"misfire_grace_s": True}, "misfire_grace_s", "宽限"),
({"overlap": "queue"}, "overlap", "skip"),
({"enabled": "yes"}, "enabled", "启用"),
({"webhook": {"format": "slack", "url": HOOK}}, "webhook", "格式"),
({"webhook": {"format": "feishu", "url": "open.feishu.cn/hook"}}, "webhook", "Webhook 地址"),
({"webhook": {"format": "feishu", "url": HOOK, "events": ["done"]}}, "webhook", "通知事件"),
({"workflow_id": ""}, "workflow_id", "工作流"),
])
async def test_invalid_create_is_422_with_field(client, over, field, fragment):
wf, _ = await workflow()
res = await client.post("/api/schedules", json=payload(wf, **over))
assert res.status_code == 422, res.text
body = res.json()
assert body["field"] == field and fragment in body["detail"], body
async def test_missing_workflow_and_unpublished_formal(client):
res = await client.post("/api/schedules", json=payload(f"gone{uuid.uuid4().hex[:6]}"))
assert res.status_code == 422 and res.json()["field"] == "workflow_id" and "不存在" in res.json()["detail"]
wf, name = await workflow(published=False)
res = await client.post("/api/schedules", json=payload(wf))
assert res.status_code == 422 and res.json()["field"] == "run_class"
assert f"「{name}」还没有发布版本" in res.json()["detail"]
# 探索运行不要求发布版本
res = await client.post("/api/schedules", json=payload(wf, run_class="exploratory"))
assert res.status_code == 201
async def test_non_json_body_is_422_in_chinese(client):
res = await client.post("/api/schedules", content=b"not json", headers={"Content-Type": "application/json"})
assert res.status_code == 422 and "JSON" in res.json()["detail"]
res = await client.post("/api/schedules", json=["列表"])
assert res.status_code == 422 and "JSON" in res.json()["detail"]
async def test_webhook_with_empty_url_on_create_means_none(client):
wf, _ = await workflow()
made = await create(client, wf, webhook={"format": "feishu", "url": "", "events": ["failed"]})
assert made["webhook"] is None
# --------------------------------------------------------------------------
# 预览、变量
# --------------------------------------------------------------------------
async def test_preview(client):
res = await client.post("/api/schedules/preview", json={
"cron": "0 9 * * *", "timezone": "Asia/Shanghai", "n": 3,
"input_template": {"stat_date": "{{ schedule.yesterday }}", "since": "{{ schedule.last_fire_time }}"}})
assert res.status_code == 200, res.text
body = res.json()
assert set(body) == {"description_text", "next", "inputs"}
assert body["description_text"] == "每天 09:00(Asia/Shanghai)"
moments = [datetime.fromisoformat(t) for t in body["next"]]
assert len(moments) == 3 and all(m.utcoffset().total_seconds() == 0 for m in moments)
assert all((b - a).total_seconds() == 86400 for a, b in zip(moments, moments[1:]))
first_local = moments[0].astimezone(ZoneInfo("Asia/Shanghai"))
assert body["inputs"]["stat_date"] == (first_local.date() - timedelta(days=1)).isoformat()
assert body["inputs"]["since"].endswith("09:00:00+08:00")
async def test_preview_defaults_to_five(client):
body = (await client.post("/api/schedules/preview", json={"cron": "*/10 * * * *", "timezone": "UTC"})).json()
assert len(body["next"]) == 5 and body["inputs"] == {}
@pytest.mark.parametrize("given, field", [
({"cron": "bad", "timezone": "UTC"}, "cron"),
({"cron": "0 9 * * *", "timezone": "Nowhere/City"}, "timezone"),
({"cron": "0 9 * * *", "timezone": "UTC", "input_template": {"d": "{{ schedule.today"}}, "input_template"),
])
async def test_preview_errors(client, given, field):
res = await client.post("/api/schedules/preview", json=given)
assert res.status_code == 422 and res.json()["field"] == field
assert any("一" <= ch <= "鿿" for ch in res.json()["detail"])
async def test_preview_checks_field_names_against_the_workflow_but_not_missing_ones(client):
wf, _ = await workflow(fields=(DATE_FIELD, {"name": "region"}))
ok = await client.post("/api/schedules/preview", json={
"cron": "0 9 * * *", "timezone": "UTC", "workflow_id": wf, "input_template": {"region": "华东"}})
assert ok.status_code == 200
bad = await client.post("/api/schedules/preview", json={
"cron": "0 9 * * *", "timezone": "UTC", "workflow_id": wf, "input_template": {"regoin": "华东"}})
assert bad.status_code == 422 and bad.json()["field"] == "input_template" and "regoin" in bad.json()["detail"]
async def test_variables(client):
res = await client.get("/api/schedules/variables")
assert res.status_code == 200
items = res.json()
assert len(items) == 10 and all(set(i) == {"name", "label", "example"} for i in items)
assert {"name": "schedule.yesterday", "label": "昨天(YYYY-MM-DD)"}.items() <= next(
i for i in items if i["name"] == "schedule.yesterday").items()
assert (await client.get("/api/schedules/variables", params={"timezone": "Nowhere/City"})).status_code == 422
# --------------------------------------------------------------------------
# 立即运行、触发历史
# --------------------------------------------------------------------------
async def test_run_now_and_overlap(client, monkeypatch):
wf, _ = await workflow()
made = await create(client, wf, webhook=None)
res = await client.post(f"/api/schedules/{made['id']}/run", headers={"X-Actor": quote("王芳")})
assert res.status_code == 200, res.text
body = res.json()
assert set(body) == {"fire", "run_id"} and set(body["fire"]) == FIRE_KEYS
fire = body["fire"]
assert (fire["kind"], fire["status"], fire["run_id"]) == ("manual", "started", body["run_id"])
assert fire["run_status"] in ("queued", "running", "succeeded")
async with SessionLocal() as session:
run = await session.get(Run, body["run_id"])
assert run.schedule_id == made["id"] and run.started_by == "王芳" and run.run_class == "formal"
assert set(run.input) == {"stat_date"}
# 上一次还没结束(这里把它改成停在审批):同一套重叠规则
await settle(body["run_id"])
async with SessionLocal() as session:
await session.execute(Run.__table__.update().where(Run.id == body["run_id"]).values(status="interrupted"))
await session.commit()
# 换到下一秒再点(同一秒内连点交回同一条,见下一个用例)
monkeypatch.setattr(schedules_api, "utcnow", lambda: datetime.now(UTC) + timedelta(seconds=2))
second = (await client.post(f"/api/schedules/{made['id']}/run")).json()
assert (second["fire"]["status"], second["run_id"]) == ("skipped_overlap", None)
assert body["run_id"][:8] in second["fire"]["reason"]
# 运行列表能按定时任务筛
listed = (await client.get("/api/runs", params={"schedule_id": made["id"]})).json()
assert [r["id"] for r in listed] == [body["run_id"]]
async def test_double_click_in_the_same_second_returns_the_same_fire(client, monkeypatch):
wf, _ = await workflow()
made = await create(client, wf, webhook=None)
moment = datetime.now(UTC).replace(microsecond=0)
monkeypatch.setattr(schedules_api, "utcnow", lambda: moment)
first, second = [(await client.post(f"/api/schedules/{made['id']}/run")).json() for _ in range(2)]
assert first["fire"]["id"] == second["fire"]["id"] and first["run_id"] == second["run_id"]
async with SessionLocal() as session:
assert await session.scalar(select(func.count()).select_from(Run).where(Run.schedule_id == made["id"])) == 1
await settle(first["run_id"])
async def test_run_now_refused_for_unpublished_formal(client):
wf, name = await workflow()
made = await create(client, wf, webhook=None)
async with SessionLocal() as session:
await session.execute(Workflow.__table__.update().where(Workflow.id == wf).values(published_version=None))
await session.commit()
body = (await client.post(f"/api/schedules/{made['id']}/run")).json()
assert (body["fire"]["status"], body["run_id"]) == ("failed_to_start", None)
assert body["fire"]["reason"] == f"「{name}」还没有发布版本,请先在编排页发布"
async def test_fire_history_pages_and_last_fire(client):
wf, _ = await workflow()
made = await create(client, wf, webhook=None)
sid = made["id"]
async with SessionLocal() as session:
for i in range(5):
session.add(ScheduleFire(schedule_id=sid, fire_at=datetime(2026, 9, 1 + i, 1, 0, tzinfo=UTC),
kind="scheduled", status="skipped_overlap", reason=f"第 {i} 次",
created_at=datetime(2026, 9, 1 + i, 1, 0, 1, tzinfo=UTC)))
await session.commit()
newest = (await session.execute(select(ScheduleFire).where(ScheduleFire.schedule_id == sid)
.order_by(ScheduleFire.created_at.desc()))).scalars().first()
session.add(WebhookDelivery(schedule_id=sid, fire_id=newest.id, event="failed", status="ok", attempts=1,
http_status=200, created_at=datetime(2026, 9, 5, 1, 0, 5, tzinfo=UTC)))
await session.commit()
page1 = (await client.get(f"/api/schedules/{sid}/fires", params={"limit": 2})).json()
assert [f["reason"] for f in page1["items"]] == ["第 4 次", "第 3 次"] and page1["next_before"]
assert page1["items"][0]["notified"] == [{"event": "failed", "status": "ok", "at": "2026-09-05T01:00:05+00:00"}]
assert page1["items"][0]["fire_at"] == "2026-09-05T01:00:00+00:00"
page2 = (await client.get(f"/api/schedules/{sid}/fires",
params={"limit": 2, "before": page1["next_before"]})).json()
assert [f["reason"] for f in page2["items"]] == ["第 2 次", "第 1 次"] and page2["next_before"]
page3 = (await client.get(f"/api/schedules/{sid}/fires",
params={"limit": 2, "before": page2["next_before"]})).json()
assert [f["reason"] for f in page3["items"]] == ["第 0 次"] and page3["next_before"] is None
last = (await client.get(f"/api/schedules/{sid}")).json()["last_fire"]
assert last["reason"] == "第 4 次" and set(last) == FIRE_KEYS
bad = await client.get(f"/api/schedules/{sid}/fires", params={"limit": "many"})
assert bad.status_code == 422 and bad.json()["field"] == "limit"
bad = await client.get(f"/api/schedules/{sid}/fires", params={"before": "昨天"})
assert bad.status_code == 422 and bad.json()["field"] == "before"
# --------------------------------------------------------------------------
# 测试 Webhook
# --------------------------------------------------------------------------
async def test_webhook_test_endpoint(client, monkeypatch):
seen: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
seen.append(request)
return httpx.Response(200, json={"code": 0, "msg": "success"})
monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(handler))
wf, _ = await workflow()
made = await create(client, wf)
res = await client.post(f"/api/schedules/{made['id']}/webhook/test")
assert res.status_code == 200
assert res.json() == {"ok": True, "http_status": 200, "message": "测试消息已送达(HTTP 200)"}
assert str(seen[0].url) == HOOK and "sign" in json.loads(seen[0].content)
bare = await create(client, wf, webhook=None)
res = await client.post(f"/api/schedules/{bare['id']}/webhook/test")
assert res.status_code == 422 and res.json()["field"] == "webhook" and "Webhook" in res.json()["detail"]