| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106 |
- """工作流回调服务 —— 流程状态变更时通知业务系统"""
- from __future__ import annotations
- import logging
- from typing import TYPE_CHECKING, Any, Callable, Dict, Optional
- import requests
- if TYPE_CHECKING:
- from app.models.models import WfProcessInstance
- logger = logging.getLogger("sqxp.workflow_callback")
- # ---------------------------------------------------------------------------
- # 本地回调注册表:内部路径(以 / 开头)直接在进程内调用,不走 HTTP
- # ---------------------------------------------------------------------------
- _LocalCallbackFn = Callable[[Any, str, str], None] # (db, business_id, event) -> None
- _LOCAL_CALLBACKS: Dict[str, _LocalCallbackFn] = {}
- def register_local_callback(path: str, fn: _LocalCallbackFn) -> None:
- """注册本地回调函数。当 callback_url 以 path 开头时直接调用 fn。"""
- _LOCAL_CALLBACKS[path] = fn
- def _dispatch_local(db: Any, instance: WfProcessInstance, event: str) -> bool:
- """尝试本地直调。返回 True 表示已处理,False 表示无匹配的本地回调。"""
- url = instance.callback_url or ""
- if not url.startswith("/"):
- return False
- for prefix, fn in _LOCAL_CALLBACKS.items():
- if url.startswith(prefix):
- try:
- fn(db, instance.business_id or "", event)
- logger.info(
- "工作流本地回调成功: event=%s instance=%s path=%s",
- event,
- instance.id,
- url,
- )
- return True
- except Exception:
- logger.exception(
- "工作流本地回调失败: event=%s instance=%s path=%s",
- event,
- instance.id,
- url,
- )
- return True # 已匹配但执行失败,不再走 HTTP
- return False
- # ---------------------------------------------------------------------------
- # trigger_callback(对外入口)
- # ---------------------------------------------------------------------------
- def trigger_callback(db: Any, instance: WfProcessInstance, event: str) -> None:
- """流程实例状态变更后,向业务系统发送回调通知。
- 优先级:
- 1. 如果 callback_url 是内部路径(以 / 开头),优先尝试本地直调
- 2. 本地无匹配时,回退到 HTTP 回调(需完整 URL)
- Args:
- db: 数据库会话。
- instance: 流程实例 ORM 对象。
- event: 事件类型,如 approved / rejected / completed。
- """
- if not instance.callback_url:
- return
- # 尝试本地直调
- if _dispatch_local(db, instance, event):
- return
- # 回退到 HTTP 回调
- payload: Dict[str, Any] = {
- "event": event,
- "instanceId": instance.id,
- "businessType": instance.business_type,
- "businessId": instance.business_id,
- "status": instance.status,
- }
- try:
- resp = requests.post(
- instance.callback_url,
- json=payload,
- timeout=10,
- headers={"Content-Type": "application/json"},
- )
- logger.info(
- "工作流HTTP回调成功: event=%s instance=%s status=%d",
- event,
- instance.id,
- resp.status_code,
- )
- except Exception:
- logger.exception(
- "工作流HTTP回调失败: event=%s instance=%s url=%s",
- event,
- instance.id,
- instance.callback_url,
- )
|