zhou
32ca8c1208
CI / Lint, Typecheck & Test (push) Successful in 1m57s
1. 新建 docs/ Sphinx 文档结构 (conf.py + 8 个 rst 章节), napoleon 支持 Google/NumPy docstring, rtd 主题, 自动生成 API 参考与错误家族文档 2. 新建 .readthedocs.yaml 配置, pyproject.toml 加 docs 依赖 3. 删除 runner.py 的 _apply_verbose_to_graph 死代码及对应测试 (功能已移入 executors.run 统一处理) 4. 更新 README: CLI 示例改为 pf 统一入口, 模块结构表补全 cli/pf.py/cli/configs/cli/_ops 等模块 5. 修复版本不一致 (pyproject.toml 0.3.5 → 0.4.5) 6. 加文档徽章链接到 ReadTheDocs
94 lines
2.7 KiB
ReStructuredText
94 lines
2.7 KiB
ReStructuredText
执行策略与 run()
|
||
=================
|
||
|
||
``run()`` 是执行入口，支持四种策略：
|
||
|
||
.. code-block:: python
|
||
|
||
report = px.run(
|
||
graph,
|
||
strategy="async", # sequential | thread | async | dependency
|
||
max_workers=8, # thread 策略的线程池大小
|
||
concurrency_limits={"db": 2}, # 按 concurrency_key 限流
|
||
dry_run=False, # True = 仅打印计划
|
||
verbose=True, # True = 打印执行过程
|
||
on_event=callback, # 状态转换回调
|
||
state=px.JSONBackend("state.json"), # 断点续跑后端
|
||
continue_on_error=False, # True = 单任务失败不中断整体
|
||
)
|
||
|
||
策略对比
|
||
--------
|
||
|
||
.. list-table::
|
||
:header-rows: 1
|
||
:widths: 18 18 30 16 18
|
||
|
||
* - 策略
|
||
- 并发模型
|
||
- 适用场景
|
||
- 同步任务
|
||
- 异步任务
|
||
* - ``sequential``
|
||
- 串行
|
||
- 调试、CPU 密集
|
||
- 直接调用
|
||
- 事件循环
|
||
* - ``thread``
|
||
- 线程池
|
||
- I/O 密集同步
|
||
- 线程池
|
||
- 不支持
|
||
* - ``async``
|
||
- 事件循环
|
||
- I/O 密集异步
|
||
- 卸载到线程池
|
||
- 事件循环
|
||
* - ``dependency``
|
||
- 依赖驱动
|
||
- 最大化并行度
|
||
- 卸载到线程池
|
||
- 事件循环
|
||
|
||
所有策略都遵循 ``RetryPolicy``、``timeout``、上下文注入、状态后端、``concurrency_limits``，
|
||
并发出 ``TaskEvent``（RUNNING/SUCCESS/FAILED/SKIPPED）。``dependency`` 策略无层屏障：
|
||
任务在其所有硬依赖完成后立即启动。
|
||
|
||
上下文注入规则
|
||
--------------
|
||
|
||
按顺序求值：
|
||
|
||
1. **标注为 ``Context``** 的参数 → 接收完整上游结果映射
|
||
2. **名称匹配依赖** 的参数 → 接收该依赖的结果（含软依赖，缺失时注入默认值）
|
||
3. **``**kwargs``** 参数 → 接收所有依赖结果（dict）
|
||
4. **``TaskSpec.args`` / ``kwargs``** → 为非依赖参数提供静态值
|
||
|
||
.. code-block:: python
|
||
|
||
from typing import Any, Dict
|
||
|
||
def aggregate(ctx: px.Context) -> Dict[str, Any]:
|
||
"""ctx 包含所有 depends_on 任务的返回值。"""
|
||
return dict(ctx)
|
||
|
||
def merge(fetch_a: str, fetch_b: str) -> str:
|
||
"""fetch_a / fetch_b 自动注入。"""
|
||
return fetch_a + fetch_b
|
||
|
||
断点续跑
|
||
--------
|
||
|
||
.. code-block:: python
|
||
|
||
from pyflowx import JSONBackend
|
||
|
||
backend = JSONBackend("state.json", ttl=3600)
|
||
report = px.run(graph, strategy="sequential", state=backend)
|
||
|
||
``run()`` 内部以 ``backend.batch()`` 包裹整个执行：所有 ``save`` 延迟到运行结束时统一落盘一次。
|
||
|
||
缓存键：默认存储键为任务名。配置 ``cache_key`` 函数后，键为 ``"name:cache_key_value"``。
|
||
|
||
完整 API 说明详见 :doc:`/api`。
|