qihongkun
d810abdcee
纳入 FastAPI 后端、Vue 管理端、MCP 桥接与文档；通过 .gitignore 排除本地数据库与构建产物。 Co-authored-by: Cursor <cursoragent@cursor.com>
954 lines
34 KiB
Python
954 lines
34 KiB
Python
import asyncio
|
||
import json
|
||
import re
|
||
import time
|
||
from collections import defaultdict, deque
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
|
||
def render_template_value(value: Any, variables: dict[str, Any]) -> Any:
|
||
if isinstance(value, str):
|
||
matches = re.findall(r"\{\{(.*?)\}\}", value)
|
||
rendered = value
|
||
for raw_key in matches:
|
||
key = raw_key.strip()
|
||
replacement = variables.get(key, "")
|
||
rendered = rendered.replace(f"{{{{{raw_key}}}}}", str(replacement))
|
||
return rendered
|
||
if isinstance(value, dict):
|
||
return {k: render_template_value(v, variables) for k, v in value.items()}
|
||
if isinstance(value, list):
|
||
return [render_template_value(v, variables) for v in value]
|
||
return value
|
||
|
||
|
||
def build_execution_order(nodes: list[dict[str, Any]], edges: list[dict[str, Any]]) -> list[str]:
|
||
node_ids = [n["id"] for n in nodes]
|
||
indegree = {node_id: 0 for node_id in node_ids}
|
||
graph = defaultdict(list)
|
||
for edge in edges:
|
||
source = edge.get("source")
|
||
target = edge.get("target")
|
||
if source in indegree and target in indegree:
|
||
graph[source].append(target)
|
||
indegree[target] += 1
|
||
|
||
queue = deque([node_id for node_id in node_ids if indegree[node_id] == 0])
|
||
order = []
|
||
while queue:
|
||
cur = queue.popleft()
|
||
order.append(cur)
|
||
for nxt in graph[cur]:
|
||
indegree[nxt] -= 1
|
||
if indegree[nxt] == 0:
|
||
queue.append(nxt)
|
||
|
||
if len(order) != len(node_ids):
|
||
return node_ids
|
||
return order
|
||
|
||
|
||
LOOP_BODY_BRANCHES = {"", "body", "loop", "in", "next", "continue"}
|
||
LOOP_EXIT_BRANCHES = {"", "done", "break", "out", "exit", "default"}
|
||
|
||
|
||
def _safe_eval(left: Any, op: str, right: Any) -> bool:
|
||
if op == "==":
|
||
return left == right
|
||
if op == "!=":
|
||
return left != right
|
||
if op == ">":
|
||
return left > right
|
||
if op == "<":
|
||
return left < right
|
||
if op == ">=":
|
||
return left >= right
|
||
if op == "<=":
|
||
return left <= right
|
||
return False
|
||
|
||
|
||
async def execute_workflow(
|
||
definition: dict[str, Any],
|
||
api_map: dict[int, dict[str, Any]],
|
||
mock_map: dict[str, dict[str, Any]],
|
||
base_url: str,
|
||
fail_fast: bool = True,
|
||
default_headers: dict[str, Any] | None = None,
|
||
on_node_start: Callable[[dict[str, Any], int, int], Any] | None = None,
|
||
on_node_finish: Callable[[dict[str, Any], dict[str, Any], int, int], Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
nodes = definition.get("nodes", [])
|
||
edges = definition.get("edges", [])
|
||
variables = dict(definition.get("variables", {}))
|
||
node_map = {node["id"]: node for node in nodes}
|
||
order = build_execution_order(nodes, edges)
|
||
order_index = {node_id: idx for idx, node_id in enumerate(order)}
|
||
results: list[dict[str, Any]] = []
|
||
halted = False
|
||
graph = _build_edge_graph(edges)
|
||
pending = deque(_sorted_start_nodes(nodes, edges, order_index))
|
||
queued = set(pending)
|
||
executed = set()
|
||
total_nodes = len(order)
|
||
default_headers = dict(default_headers or {})
|
||
|
||
async with httpx.AsyncClient(timeout=20.0) as client:
|
||
while pending:
|
||
node_id = pending.popleft()
|
||
queued.discard(node_id)
|
||
if node_id in executed or node_id not in node_map:
|
||
continue
|
||
node = node_map[node_id]
|
||
node_type = str(node.get("data", {}).get("type", "http")).lower()
|
||
await _invoke_callback(on_node_start, node, len(executed) + 1, total_nodes)
|
||
|
||
if node_type == "loop":
|
||
loop_results, result = await _execute_loop_node(
|
||
node=node,
|
||
node_map=node_map,
|
||
graph=graph,
|
||
edges=edges,
|
||
order_index=order_index,
|
||
variables=variables,
|
||
api_map=api_map,
|
||
mock_map=mock_map,
|
||
base_url=base_url,
|
||
client=client,
|
||
default_headers=default_headers,
|
||
fail_fast=fail_fast,
|
||
on_node_start=on_node_start,
|
||
on_node_finish=on_node_finish,
|
||
executed_count=len(executed),
|
||
total_nodes=total_nodes,
|
||
)
|
||
results.extend(loop_results)
|
||
else:
|
||
result = await execute_single_node(
|
||
node, variables, api_map, mock_map, base_url, client, default_headers=default_headers
|
||
)
|
||
results.append(result)
|
||
|
||
executed.add(node_id)
|
||
_persist_node_last_run(node, result)
|
||
await _invoke_callback(on_node_finish, node, result, len(executed), total_nodes)
|
||
|
||
if result["status"] == "failed" and fail_fast:
|
||
halted = True
|
||
break
|
||
|
||
allowed = _allowed_branches(result)
|
||
next_nodes = []
|
||
for target, branch in graph.get(node_id, []):
|
||
if target in executed or target in queued:
|
||
continue
|
||
if node_type == "loop":
|
||
if branch not in LOOP_EXIT_BRANCHES:
|
||
continue
|
||
if branch in allowed:
|
||
next_nodes.append(target)
|
||
next_nodes.sort(key=lambda item: order_index.get(item, 10**9))
|
||
for target in next_nodes:
|
||
pending.append(target)
|
||
queued.add(target)
|
||
|
||
if halted:
|
||
for node_id in order:
|
||
if node_id in executed:
|
||
continue
|
||
node_type = node_map[node_id].get("data", {}).get("type", "http")
|
||
skipped_result = {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "skipped",
|
||
"output": {},
|
||
"error": "halted by previous failure",
|
||
}
|
||
results.append(skipped_result)
|
||
_persist_node_last_run(node_map[node_id], skipped_result)
|
||
await _invoke_callback(on_node_finish, node_map[node_id], skipped_result, len(executed), total_nodes)
|
||
|
||
status = "failed" if any(item["status"] == "failed" for item in results) else "success"
|
||
return {"status": status, "variables": variables, "results": results, "raw": json.dumps(results)}
|
||
|
||
|
||
async def _invoke_callback(callback: Callable[..., Any] | None, *args: Any) -> None:
|
||
if callback is None:
|
||
return
|
||
result = callback(*args)
|
||
if asyncio.iscoroutine(result):
|
||
await result
|
||
|
||
|
||
def _persist_node_last_run(node: dict[str, Any], result: dict[str, Any]) -> None:
|
||
if not isinstance(node, dict):
|
||
return
|
||
data = node.setdefault("data", {})
|
||
if not isinstance(data, dict):
|
||
return
|
||
data["last_run"] = {
|
||
"status": result.get("status"),
|
||
"error": result.get("error"),
|
||
"output": result.get("output", {}),
|
||
"ts": int(time.time()),
|
||
}
|
||
|
||
|
||
def _edge_branch(edge: dict[str, Any]) -> str:
|
||
data = edge.get("data")
|
||
if isinstance(data, dict):
|
||
value = str(data.get("branch", "")).strip().lower()
|
||
if value:
|
||
return value
|
||
label = str(edge.get("label", "")).strip().lower()
|
||
return label
|
||
|
||
|
||
def _build_edge_graph(edges: list[dict[str, Any]]) -> dict[str, list[tuple[str, str]]]:
|
||
graph: dict[str, list[tuple[str, str]]] = defaultdict(list)
|
||
for edge in edges:
|
||
source = edge.get("source")
|
||
target = edge.get("target")
|
||
if not source or not target:
|
||
continue
|
||
graph[source].append((target, _edge_branch(edge)))
|
||
return graph
|
||
|
||
|
||
def _sorted_start_nodes(
|
||
nodes: list[dict[str, Any]], edges: list[dict[str, Any]], order_index: dict[str, int]
|
||
) -> list[str]:
|
||
indegree = {node["id"]: 0 for node in nodes if "id" in node}
|
||
for edge in edges:
|
||
target = edge.get("target")
|
||
if target in indegree:
|
||
indegree[target] += 1
|
||
starts = [node_id for node_id, degree in indegree.items() if degree == 0]
|
||
starts.sort(key=lambda item: order_index.get(item, 10**9))
|
||
return starts
|
||
|
||
|
||
def _allowed_branches(result: dict[str, Any]) -> set[str]:
|
||
status = result.get("status")
|
||
node_type = result.get("node_type")
|
||
output = result.get("output", {})
|
||
if status == "failed":
|
||
return {"", "always", "default", "fail", "failed", "failure", "on_fail"}
|
||
if node_type == "condition":
|
||
if bool(output.get("result")):
|
||
return {"", "always", "default", "true", "on_true", "success"}
|
||
return {"", "always", "default", "false", "on_false"}
|
||
if node_type == "loop":
|
||
return set(LOOP_EXIT_BRANCHES) | {"always", "success", "on_success"}
|
||
return {"", "always", "default", "success", "on_success"}
|
||
|
||
|
||
def _resolve_field_path(source: Any, field: str) -> Any:
|
||
extracted = source
|
||
for part in str(field or "").split("."):
|
||
if not part:
|
||
continue
|
||
if isinstance(extracted, dict):
|
||
extracted = extracted.get(part)
|
||
else:
|
||
return None
|
||
return extracted
|
||
|
||
|
||
def _resolve_condition_operand(node_data: dict[str, Any], side: str, variables: dict[str, Any]) -> Any:
|
||
mode_key = f"{side}_mode"
|
||
mode = str(node_data.get(mode_key) or node_data.get("mode", "template")).strip().lower()
|
||
if mode == "json_path":
|
||
source_var = str(node_data.get(f"{side}_source_var") or node_data.get("source_var") or "").strip()
|
||
field = str(node_data.get(f"{side}_field") or node_data.get("field") or "").strip()
|
||
if not source_var:
|
||
return None
|
||
return _resolve_field_path(variables.get(source_var), field)
|
||
|
||
raw = node_data.get(side, "")
|
||
return render_template_value(raw, variables)
|
||
|
||
|
||
def _evaluate_condition(node_data: dict[str, Any], variables: dict[str, Any]) -> bool:
|
||
left = _resolve_condition_operand(node_data, "left", variables)
|
||
right = _resolve_condition_operand(node_data, "right", variables)
|
||
op = str(node_data.get("op", "=="))
|
||
return _safe_eval(left, op, right)
|
||
|
||
|
||
def _collect_loop_body_node_ids(loop_id: str, graph: dict[str, list[tuple[str, str]]]) -> set[str]:
|
||
entries = [target for target, branch in graph.get(loop_id, []) if branch in LOOP_BODY_BRANCHES]
|
||
if not entries:
|
||
return set()
|
||
|
||
body: set[str] = set()
|
||
queue = deque(entries)
|
||
while queue:
|
||
current = queue.popleft()
|
||
if current == loop_id or current in body:
|
||
continue
|
||
body.add(current)
|
||
for target, _branch in graph.get(current, []):
|
||
if target != loop_id and target not in body:
|
||
queue.append(target)
|
||
return body
|
||
|
||
|
||
async def _execute_node_subgraph(
|
||
*,
|
||
node_ids: set[str],
|
||
entry_node_ids: list[str],
|
||
node_map: dict[str, dict[str, Any]],
|
||
graph: dict[str, list[tuple[str, str]]],
|
||
edges: list[dict[str, Any]],
|
||
variables: dict[str, Any],
|
||
api_map: dict[int, dict[str, Any]],
|
||
mock_map: dict[str, dict[str, Any]],
|
||
base_url: str,
|
||
client: httpx.AsyncClient,
|
||
default_headers: dict[str, Any],
|
||
fail_fast: bool,
|
||
iteration: int,
|
||
on_node_start: Callable[[dict[str, Any], int, int], Any] | None,
|
||
on_node_finish: Callable[[dict[str, Any], dict[str, Any], int, int], Any] | None,
|
||
executed_count: int,
|
||
total_nodes: int,
|
||
) -> tuple[list[dict[str, Any]], bool]:
|
||
if not node_ids:
|
||
return [], False
|
||
|
||
subgraph_nodes = [node_map[node_id] for node_id in node_ids if node_id in node_map]
|
||
subgraph_edges = [
|
||
edge
|
||
for edge in edges
|
||
if edge.get("source") in node_ids and edge.get("target") in node_ids
|
||
]
|
||
order = build_execution_order(subgraph_nodes, subgraph_edges)
|
||
order_index = {node_id: idx for idx, node_id in enumerate(order)}
|
||
|
||
indegree = {node_id: 0 for node_id in node_ids}
|
||
for edge in subgraph_edges:
|
||
target = edge.get("target")
|
||
if target in indegree:
|
||
indegree[target] += 1
|
||
|
||
starts = [node_id for node_id in entry_node_ids if node_id in node_ids]
|
||
if not starts:
|
||
starts = [node_id for node_id, degree in indegree.items() if degree == 0]
|
||
starts.sort(key=lambda item: order_index.get(item, 10**9))
|
||
|
||
pending = deque(starts)
|
||
queued = set(starts)
|
||
local_executed: set[str] = set()
|
||
results: list[dict[str, Any]] = []
|
||
halted = False
|
||
|
||
while pending:
|
||
node_id = pending.popleft()
|
||
queued.discard(node_id)
|
||
if node_id in local_executed or node_id not in node_map:
|
||
continue
|
||
|
||
node = node_map[node_id]
|
||
await _invoke_callback(on_node_start, node, executed_count + len(local_executed) + 1, total_nodes)
|
||
result = await execute_single_node(
|
||
node, variables, api_map, mock_map, base_url, client, default_headers=default_headers
|
||
)
|
||
if isinstance(result.get("output"), dict):
|
||
result["output"]["loop_iteration"] = iteration
|
||
results.append(result)
|
||
local_executed.add(node_id)
|
||
_persist_node_last_run(node, result)
|
||
await _invoke_callback(on_node_finish, node, result, executed_count + len(local_executed), total_nodes)
|
||
|
||
if result["status"] == "failed" and fail_fast:
|
||
halted = True
|
||
break
|
||
|
||
allowed = _allowed_branches(result)
|
||
next_nodes = []
|
||
for target, branch in graph.get(node_id, []):
|
||
if target not in node_ids or target in local_executed or target in queued:
|
||
continue
|
||
if branch in allowed:
|
||
next_nodes.append(target)
|
||
next_nodes.sort(key=lambda item: order_index.get(item, 10**9))
|
||
for target in next_nodes:
|
||
pending.append(target)
|
||
queued.add(target)
|
||
|
||
return results, halted
|
||
|
||
|
||
async def _execute_loop_node(
|
||
*,
|
||
node: dict[str, Any],
|
||
node_map: dict[str, dict[str, Any]],
|
||
graph: dict[str, list[tuple[str, str]]],
|
||
edges: list[dict[str, Any]],
|
||
order_index: dict[str, int],
|
||
variables: dict[str, Any],
|
||
api_map: dict[int, dict[str, Any]],
|
||
mock_map: dict[str, dict[str, Any]],
|
||
base_url: str,
|
||
client: httpx.AsyncClient,
|
||
default_headers: dict[str, Any],
|
||
fail_fast: bool,
|
||
on_node_start: Callable[[dict[str, Any], int, int], Any] | None,
|
||
on_node_finish: Callable[[dict[str, Any], dict[str, Any], int, int], Any] | None,
|
||
executed_count: int,
|
||
total_nodes: int,
|
||
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||
node_id = str(node.get("id", ""))
|
||
node_data = node.get("data", {}) if isinstance(node.get("data"), dict) else {}
|
||
max_iterations = max(1, min(int(node_data.get("max_iterations", 10) or 10), 1000))
|
||
iteration_var = str(node_data.get("iteration_var") or "loop_index").strip() or "loop_index"
|
||
|
||
body_node_ids = _collect_loop_body_node_ids(node_id, graph)
|
||
entry_node_ids = [
|
||
target for target, branch in graph.get(node_id, []) if branch in LOOP_BODY_BRANCHES and target in body_node_ids
|
||
]
|
||
entry_node_ids.sort(key=lambda item: order_index.get(item, 10**9))
|
||
|
||
if not body_node_ids:
|
||
return [], {
|
||
"node_id": node_id,
|
||
"node_type": "loop",
|
||
"status": "failed",
|
||
"output": {"iterations": 0, "reason": "missing loop body branch"},
|
||
"error": "循环节点需要连接 body 分支到循环体",
|
||
}
|
||
|
||
all_results: list[dict[str, Any]] = []
|
||
iterations_run = 0
|
||
continue_loop = True
|
||
last_error: str | None = None
|
||
|
||
while continue_loop and iterations_run < max_iterations:
|
||
variables[iteration_var] = iterations_run
|
||
body_results, halted = await _execute_node_subgraph(
|
||
node_ids=body_node_ids,
|
||
entry_node_ids=entry_node_ids,
|
||
node_map=node_map,
|
||
graph=graph,
|
||
edges=edges,
|
||
variables=variables,
|
||
api_map=api_map,
|
||
mock_map=mock_map,
|
||
base_url=base_url,
|
||
client=client,
|
||
default_headers=default_headers,
|
||
fail_fast=fail_fast,
|
||
iteration=iterations_run,
|
||
on_node_start=on_node_start,
|
||
on_node_finish=on_node_finish,
|
||
executed_count=executed_count + len(all_results),
|
||
total_nodes=total_nodes,
|
||
)
|
||
all_results.extend(body_results)
|
||
iterations_run += 1
|
||
|
||
if halted or any(item.get("status") == "failed" for item in body_results):
|
||
last_error = next(
|
||
(str(item.get("error") or "loop body failed") for item in body_results if item.get("status") == "failed"),
|
||
"loop body failed",
|
||
)
|
||
break
|
||
|
||
while_cfg = {
|
||
"left_mode": node_data.get("while_left_mode") or node_data.get("left_mode"),
|
||
"left": node_data.get("while_left", ""),
|
||
"left_source_var": node_data.get("while_left_source_var") or node_data.get("left_source_var"),
|
||
"left_field": node_data.get("while_left_field") or node_data.get("left_field"),
|
||
"op": node_data.get("while_op", node_data.get("op", "==")),
|
||
"right_mode": node_data.get("while_right_mode") or node_data.get("right_mode"),
|
||
"right": node_data.get("while_right", ""),
|
||
"right_source_var": node_data.get("while_right_source_var"),
|
||
"right_field": node_data.get("while_right_field"),
|
||
}
|
||
has_while_rule = any(
|
||
str(while_cfg.get(key) or "").strip()
|
||
for key in ("left", "left_source_var", "left_field", "right", "right_source_var", "right_field")
|
||
)
|
||
if not has_while_rule:
|
||
continue_loop = False
|
||
else:
|
||
continue_loop = _evaluate_condition(while_cfg, variables)
|
||
|
||
save_as = str(node_data.get("save_as") or f"{node_id}_loop").strip()
|
||
summary = {
|
||
"iterations": iterations_run,
|
||
"max_iterations": max_iterations,
|
||
"continue_loop": continue_loop,
|
||
"body_nodes": sorted(body_node_ids),
|
||
}
|
||
variables[save_as] = summary
|
||
|
||
status = "failed" if last_error else "success"
|
||
return all_results, {
|
||
"node_id": node_id,
|
||
"node_type": "loop",
|
||
"status": status,
|
||
"output": summary,
|
||
"error": last_error,
|
||
}
|
||
|
||
|
||
async def execute_single_node(
|
||
node: dict[str, Any],
|
||
variables: dict[str, Any],
|
||
api_map: dict[int, dict[str, Any]],
|
||
mock_map: dict[str, dict[str, Any]],
|
||
base_url: str,
|
||
client: httpx.AsyncClient | None = None,
|
||
default_headers: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
node_id = node.get("id", "")
|
||
node_data = node.get("data", {})
|
||
node_type = node_data.get("type", "http")
|
||
own_client = client is None
|
||
if own_client:
|
||
client = httpx.AsyncClient(timeout=20.0)
|
||
assert client is not None
|
||
try:
|
||
# Legacy compatibility: 旧的 mock 节点已废弃 → 视为 http
|
||
if node_type == "mock":
|
||
node_type = "http"
|
||
node_data["type"] = "http"
|
||
|
||
if node_type == "http":
|
||
return await _execute_http_node(
|
||
node_id=node_id,
|
||
node_data=node_data,
|
||
variables=variables,
|
||
api_map=api_map,
|
||
mock_map=mock_map,
|
||
base_url=base_url,
|
||
client=client,
|
||
default_headers=default_headers or {},
|
||
)
|
||
|
||
if node_type == "condition":
|
||
ok = _evaluate_condition(node_data, variables)
|
||
variables[node_data.get("save_as", f"{node_id}_result")] = ok
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": {
|
||
"result": ok,
|
||
"left": _resolve_condition_operand(node_data, "left", variables),
|
||
"right": _resolve_condition_operand(node_data, "right", variables),
|
||
"op": node_data.get("op", "=="),
|
||
},
|
||
"error": None,
|
||
}
|
||
|
||
if node_type == "loop":
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "failed",
|
||
"output": {},
|
||
"error": "loop node must be executed by execute_workflow",
|
||
}
|
||
|
||
if node_type in {"start", "end"}:
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": {"pass": True},
|
||
"error": None,
|
||
}
|
||
|
||
if node_type == "extract":
|
||
source_key = node_data.get("source_var", "")
|
||
field = node_data.get("field", "")
|
||
save_as = node_data.get("save_as", "")
|
||
source = variables.get(source_key, {})
|
||
extracted = source
|
||
for part in field.split("."):
|
||
if not part:
|
||
continue
|
||
if isinstance(extracted, dict):
|
||
extracted = extracted.get(part)
|
||
else:
|
||
extracted = None
|
||
break
|
||
variables[save_as] = extracted
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": {"value": extracted},
|
||
"error": None,
|
||
}
|
||
|
||
if node_type == "transform":
|
||
template = node_data.get("template", {})
|
||
transformed = render_template_value(template, variables)
|
||
save_as = node_data.get("save_as", f"{node_id}_output")
|
||
variables[save_as] = transformed
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": {"value": transformed},
|
||
"error": None,
|
||
}
|
||
|
||
if node_type == "delay":
|
||
seconds = float(node_data.get("seconds", 1))
|
||
await asyncio.sleep(seconds)
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": {"waited": seconds},
|
||
"error": None,
|
||
}
|
||
|
||
if node_type == "ai":
|
||
prompt = node_data.get("prompt", "")
|
||
response = {"message": "AI node reserved", "prompt": prompt}
|
||
save_as = node_data.get("save_as", f"{node_id}_ai")
|
||
variables[save_as] = response
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "success",
|
||
"output": response,
|
||
"error": None,
|
||
}
|
||
|
||
raise ValueError(f"unsupported node type: {node_type}")
|
||
except Exception as exc:
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": node_type,
|
||
"status": "failed",
|
||
"output": {},
|
||
"error": str(exc),
|
||
}
|
||
finally:
|
||
if own_client:
|
||
await client.aclose()
|
||
|
||
|
||
def _merge_str_dict(a: dict[str, Any] | None, b: dict[str, Any] | None) -> dict[str, Any]:
|
||
out = dict(a or {})
|
||
for k, v in (b or {}).items():
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def _mock_payload_for_overlay(
|
||
node_data: dict[str, Any], mock_map: dict[str, dict[str, Any]], variables: dict[str, Any]
|
||
) -> dict[str, Any]:
|
||
raw: Any = node_data.get("mock_data")
|
||
if raw is None:
|
||
raw = node_data.get("mock")
|
||
if isinstance(raw, str) and raw in mock_map:
|
||
raw = mock_map.get(raw) or {}
|
||
name = node_data.get("mock_name")
|
||
if (not raw or raw == {}) and isinstance(name, str) and name.strip() and name in mock_map:
|
||
raw = mock_map.get(name) or {}
|
||
rendered = render_template_value(raw if isinstance(raw, dict) else {}, variables)
|
||
return rendered if isinstance(rendered, dict) else {}
|
||
|
||
|
||
def _looks_like_response_mock(m: dict[str, Any]) -> bool:
|
||
if not m:
|
||
return False
|
||
if "status_code" in m:
|
||
return True
|
||
if m.get("text") is not None or m.get("json") is not None:
|
||
if not any(k in m for k in ("method", "url", "query", "path_params")):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _has_request_overlay_keys(m: dict[str, Any]) -> bool:
|
||
return any(k in m for k in ("method", "url", "headers", "query", "body", "path_params"))
|
||
|
||
|
||
def _apply_request_overlay(
|
||
method: str,
|
||
url: str,
|
||
headers: dict[str, Any],
|
||
params: dict[str, Any],
|
||
json_body: dict[str, Any],
|
||
path_params: dict[str, Any],
|
||
overlay: dict[str, Any],
|
||
) -> tuple[str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any]]:
|
||
if not overlay:
|
||
return method, url, headers, params, json_body, path_params
|
||
if overlay.get("method"):
|
||
method = str(overlay["method"]).upper()
|
||
if "url" in overlay and overlay["url"] is not None:
|
||
url = str(overlay["url"])
|
||
if isinstance(overlay.get("headers"), dict):
|
||
headers = _merge_str_dict(headers, overlay["headers"])
|
||
if isinstance(overlay.get("query"), dict):
|
||
params = dict(overlay["query"])
|
||
if isinstance(overlay.get("body"), dict):
|
||
json_body = dict(overlay["body"])
|
||
if isinstance(overlay.get("path_params"), dict):
|
||
path_params = dict(overlay["path_params"])
|
||
return method, url, headers, params, json_body, path_params
|
||
|
||
|
||
async def _execute_http_node(
|
||
node_id: str,
|
||
node_data: dict[str, Any],
|
||
variables: dict[str, Any],
|
||
api_map: dict[int, dict[str, Any]],
|
||
mock_map: dict[str, dict[str, Any]],
|
||
base_url: str,
|
||
client: httpx.AsyncClient,
|
||
default_headers: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
request_cfg = _resolve_http_request(node_data, api_map)
|
||
mode = str(node_data.get("mode", "http")).lower()
|
||
overlay_cfg = _mock_payload_for_overlay(node_data, mock_map, variables)
|
||
|
||
use_request_mock = bool(node_data.get("use_mock"))
|
||
legacy_mode_mock = mode == "mock"
|
||
# 旧 mode=mock：若 overlay 中含请求字段则作为请求 mock；仅有响应字段时不覆盖请求（仍发真实 HTTP）
|
||
if legacy_mode_mock:
|
||
use_request_mock = use_request_mock or _has_request_overlay_keys(overlay_cfg)
|
||
|
||
method = str(request_cfg.get("method", "GET")).upper()
|
||
url = render_template_value(request_cfg.get("url", ""), variables)
|
||
|
||
dh = render_template_value(dict(default_headers or {}), variables) or {}
|
||
base_h = render_template_value(request_cfg.get("headers", {}), variables) or {}
|
||
headers = _merge_str_dict(dh, base_h)
|
||
params = render_template_value(request_cfg.get("query", {}), variables) or {}
|
||
json_body = render_template_value(request_cfg.get("body", {}), variables) or {}
|
||
path_params = render_template_value(request_cfg.get("path_params", {}), variables) or {}
|
||
|
||
mock_request_applied = False
|
||
if use_request_mock and overlay_cfg and _has_request_overlay_keys(overlay_cfg):
|
||
method, url, headers, params, json_body, path_params = _apply_request_overlay(
|
||
method, url, headers, params, json_body, path_params, overlay_cfg
|
||
)
|
||
mock_request_applied = True
|
||
|
||
if base_url and url and not url.startswith("http"):
|
||
url = f"{base_url.rstrip('/')}/{url.lstrip('/')}"
|
||
|
||
for key, value in path_params.items():
|
||
url = url.replace(f"{{{key}}}", str(value))
|
||
|
||
request_payload = {
|
||
"method": method,
|
||
"url": url,
|
||
"headers": headers,
|
||
"query": params,
|
||
"body": json_body if method in {"POST", "PUT", "PATCH", "DELETE"} else None,
|
||
"path_params": path_params,
|
||
"mock_request_applied": mock_request_applied,
|
||
}
|
||
|
||
response_payload: dict[str, Any] = {}
|
||
error_message: str | None = None
|
||
mock_response_used = False
|
||
started = time.perf_counter()
|
||
|
||
try:
|
||
response = await client.request(
|
||
method=method,
|
||
url=url,
|
||
headers=headers,
|
||
params=params,
|
||
json=json_body if method in {"POST", "PUT", "PATCH", "DELETE"} else None,
|
||
)
|
||
response_payload = _build_response_payload(response)
|
||
except Exception as exc:
|
||
error_message = str(exc)
|
||
if mode == "auto" and overlay_cfg and _looks_like_response_mock(overlay_cfg):
|
||
response_payload = _normalize_mock_response(overlay_cfg)
|
||
response_payload["mock_fallback_reason"] = error_message
|
||
mock_response_used = True
|
||
error_message = None
|
||
|
||
duration_ms = int((time.perf_counter() - started) * 1000)
|
||
assertions = _evaluate_http_assertions(node_data.get("expect"), response_payload, error_message)
|
||
status = "failed" if (error_message or any(item["passed"] is False for item in assertions)) else "success"
|
||
|
||
output = {
|
||
"request": request_payload,
|
||
"response": response_payload,
|
||
"mock_used": mock_request_applied or mock_response_used,
|
||
"duration_ms": duration_ms,
|
||
"assertions": assertions,
|
||
}
|
||
|
||
save_as = node_data.get("save_as")
|
||
if save_as:
|
||
variables[save_as] = response_payload
|
||
|
||
if status == "failed" and error_message is None:
|
||
error_message = "; ".join(item.get("reason", "assertion failed") for item in assertions if item["passed"] is False)
|
||
|
||
return {
|
||
"node_id": node_id,
|
||
"node_type": "http",
|
||
"status": status,
|
||
"output": output,
|
||
"error": error_message,
|
||
}
|
||
|
||
|
||
def _resolve_http_request(node_data: dict[str, Any], api_map: dict[int, dict[str, Any]]) -> dict[str, Any]:
|
||
"""Merge api_map definition (if api_id given) with inline node fields. Inline overrides."""
|
||
|
||
base: dict[str, Any] = {}
|
||
api_id_value = node_data.get("api_id")
|
||
if api_id_value is not None:
|
||
try:
|
||
api_id = int(api_id_value)
|
||
except (TypeError, ValueError):
|
||
api_id = None
|
||
if api_id is not None and api_id in api_map:
|
||
api_def = api_map[api_id]
|
||
base = {
|
||
"method": api_def.get("method", "GET"),
|
||
"url": api_def.get("url", ""),
|
||
"headers": dict(api_def.get("headers", {}) or {}),
|
||
"query": dict(api_def.get("query", {}) or {}),
|
||
"body": dict(api_def.get("body", {}) or {}),
|
||
"path_params": dict(api_def.get("path_params", {}) or {}),
|
||
}
|
||
|
||
for field in ("method", "url"):
|
||
if node_data.get(field):
|
||
base[field] = node_data[field]
|
||
for field in ("headers", "query", "body", "path_params"):
|
||
if isinstance(node_data.get(field), dict):
|
||
merged = dict(base.get(field, {}) or {})
|
||
merged.update(node_data[field])
|
||
base[field] = merged
|
||
|
||
base.setdefault("method", "GET")
|
||
base.setdefault("url", "")
|
||
base.setdefault("headers", {})
|
||
base.setdefault("query", {})
|
||
base.setdefault("body", {})
|
||
base.setdefault("path_params", {})
|
||
return base
|
||
|
||
|
||
def _build_response_payload(response: httpx.Response) -> dict[str, Any]:
|
||
payload: dict[str, Any] = {
|
||
"status_code": response.status_code,
|
||
"headers": dict(response.headers),
|
||
"text": response.text,
|
||
}
|
||
try:
|
||
payload["json"] = response.json()
|
||
except Exception:
|
||
payload["json"] = None
|
||
return payload
|
||
|
||
|
||
def _normalize_mock_response(mock_data: dict[str, Any]) -> dict[str, Any]:
|
||
if not isinstance(mock_data, dict):
|
||
return {"status_code": 200, "headers": {}, "text": "", "json": None}
|
||
body = mock_data.get("json")
|
||
if body is None and "data" in mock_data:
|
||
body = mock_data.get("data")
|
||
text = mock_data.get("text")
|
||
if text is None and body is not None:
|
||
try:
|
||
text = json.dumps(body, ensure_ascii=False)
|
||
except Exception:
|
||
text = str(body)
|
||
return {
|
||
"status_code": int(mock_data.get("status_code", 200)),
|
||
"headers": dict(mock_data.get("headers", {}) or {}),
|
||
"text": text or "",
|
||
"json": body,
|
||
"is_mock": True,
|
||
}
|
||
|
||
|
||
def _evaluate_http_assertions(
|
||
expect: Any, response_payload: dict[str, Any], error_message: str | None
|
||
) -> list[dict[str, Any]]:
|
||
if not isinstance(expect, dict):
|
||
return []
|
||
if error_message:
|
||
return [{"name": "request", "passed": False, "reason": error_message}]
|
||
|
||
results: list[dict[str, Any]] = []
|
||
status_in = expect.get("status_in") or expect.get("status_codes")
|
||
if isinstance(status_in, list) and status_in:
|
||
actual = response_payload.get("status_code")
|
||
passed = actual in status_in
|
||
results.append(
|
||
{
|
||
"name": "status_in",
|
||
"passed": passed,
|
||
"expected": status_in,
|
||
"actual": actual,
|
||
"reason": None if passed else f"status_code {actual} not in {status_in}",
|
||
}
|
||
)
|
||
|
||
json_contains = expect.get("json_contains")
|
||
if isinstance(json_contains, dict):
|
||
body = response_payload.get("json") or {}
|
||
passed, reason = _dict_contains(body, json_contains)
|
||
results.append(
|
||
{
|
||
"name": "json_contains",
|
||
"passed": passed,
|
||
"expected": json_contains,
|
||
"actual": body,
|
||
"reason": None if passed else reason,
|
||
}
|
||
)
|
||
|
||
text_includes = expect.get("text_includes")
|
||
if isinstance(text_includes, str):
|
||
text = response_payload.get("text") or ""
|
||
passed = text_includes in text
|
||
results.append(
|
||
{
|
||
"name": "text_includes",
|
||
"passed": passed,
|
||
"expected": text_includes,
|
||
"actual": text[:200],
|
||
"reason": None if passed else f"response text does not contain '{text_includes}'",
|
||
}
|
||
)
|
||
|
||
return results
|
||
|
||
|
||
def _dict_contains(actual: Any, expected: Any) -> tuple[bool, str | None]:
|
||
if isinstance(expected, dict):
|
||
if not isinstance(actual, dict):
|
||
return False, "expected object"
|
||
for key, value in expected.items():
|
||
if key not in actual:
|
||
return False, f"missing key: {key}"
|
||
ok, reason = _dict_contains(actual[key], value)
|
||
if not ok:
|
||
return False, f"{key}: {reason}"
|
||
return True, None
|
||
if isinstance(expected, list):
|
||
if not isinstance(actual, list):
|
||
return False, "expected list"
|
||
for item in expected:
|
||
if item not in actual:
|
||
return False, f"missing list item: {item}"
|
||
return True, None
|
||
return (actual == expected, None if actual == expected else f"expected {expected}, got {actual}")
|