HsxWorkFlow 提供了基于 FastAPI + Socket.IO 的 Web 管理界面，支持 RESTful API 和 WebSocket 实时通信。本文档详细介绍 Web 管理界面的使用方法。
from hsxworkflow import WorkflowManager, WorkRegisterHandler
wf = WorkRegisterHandler(step_key="demo")
@wf.register_class()
class DemoWorkFlow(StepActionHandler):
@wf.step(sort=1)
def hello(self):
return self.set_success()
manager = WorkflowManager()
manager.register_object(wf)
manager.run_app() # 访问 http://localhost:5050
HsxWorkFlow 基于 APScheduler 提供强大的定时调度功能，支持 Cron 表达式、间隔执行和一次性任务。
WorkflowScheduler 是工作流调度器的核心类，封装了 APScheduler 的功能。
from hsxworkflow import WorkflowManager
from hsxworkflow.scheduler import WorkflowScheduler
manager = WorkflowManager()
manager.register_object(wf)
scheduler = WorkflowScheduler(manager)
HsxWorkFlow 提供了许多高级特性，包括步骤重试、子步骤分支、参数传递、全局数据共享、生命周期钩子等。本文档详细介绍这些高级特性的使用方法。
步骤执行失败时自动重试，可配置重试次数和重试间隔。
@wf.step(sort=1, retry=3, retry_interval=2)
def unreliable_step(self):
"""最多重试 3 次，每次间隔 2 秒"""
result = call_external_api()
if result.ok:
return self.set_success(data=result.data)
return self.set_failure(message="API 调用失败")