zhou
cd38e1246a
变更包括： 1. 更新pyproject.toml行长度限制为120 2. 简化多处异常提示字符串的换行写法 3. 批量使用Any类型泛型优化类型标注 4. 重构cli/pymake.py的配置与任务定义 5. 删除冗余的测试代码与废弃的pymake测试文件 6. 修复示例代码的类型注解
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
"""Example 3: async aggregation with static args and Context injection.
|
|
|
|
Shows:
|
|
* async task functions executed with strategy="async".
|
|
* static positional args (TaskSpec.args) for parameterised tasks.
|
|
* Context annotation to receive the full upstream result mapping.
|
|
* on_event callback for real-time progress.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
import pyflowx as px
|
|
|
|
|
|
async def fetch_user(uid: int) -> dict[str, Any]:
|
|
await asyncio.sleep(0.2)
|
|
return {"id": uid, "name": f"User{uid}"}
|
|
|
|
|
|
async def fetch_posts(uid: int) -> list[int]:
|
|
await asyncio.sleep(0.2)
|
|
return [uid, uid + 1]
|
|
|
|
|
|
# Context annotation → receives the full mapping of upstream results.
|
|
def aggregate(ctx: px.Context) -> dict[str, Any]:
|
|
return dict(ctx)
|
|
|
|
|
|
def main() -> None:
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
# Static positional args parameterise the same function twice.
|
|
px.TaskSpec("fetch_user", fetch_user, args=(1,)),
|
|
px.TaskSpec("fetch_posts", fetch_posts, args=(1,)),
|
|
px.TaskSpec("aggregate", aggregate, depends_on=("fetch_user", "fetch_posts")),
|
|
]
|
|
)
|
|
|
|
print("=== Dry run ===")
|
|
_ = px.run(graph, strategy="async", dry_run=True)
|
|
|
|
events: list[px.TaskEvent] = []
|
|
print("\n=== Async execution ===")
|
|
report = px.run(graph, strategy="async", on_event=events.append)
|
|
|
|
for ev in events:
|
|
print(f" event: {ev.task} -> {ev.status.value}")
|
|
|
|
print(f"\naggregate = {report['aggregate']}")
|
|
print(report.describe())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|