fasttask_manager

Manager for fasttask

Installation

pip install fasttask_manager            # 同步 Manager（只依赖 retry / requests）
pip install fasttask_manager[async]     # 需要 AsyncManager 时（额外装 httpx / tenacity）

Usage

Create a Manager

Synchronous Manager:

from fasttask_manager import Manager

m = Manager("127.0.0.1", port=8080, protocol="https")

Asynchronous Manager:

（需要异步依赖：pip install fasttask_manager[async]）

import asyncio
from fasttask_manager import AsyncManager

async def main():
    am = AsyncManager("127.0.0.1", port=8080, protocol="https")
    # use await
    resp = await am.run("get_hypotenuse", {"a": 3, "b": 4})

Run a task

resp = m.run("get_hypotenuse", {"a": 3, "b": 4})

Create a task and check result later

result_id = m.create_task("get_hypotenuse", {"a": 3, "b": 4})["id"]
# do something...
resp = m.check("get_hypotenuse", result_id)   # 完整响应；外置结果已自动下载
result = resp["result"]

Create a task and wait for result

result = m.create_and_wait_result("get_hypotenuse", {"a": 3, "b": 4})

Large results (offloaded to object storage)

当 FastTask 配置了 RESULT_TYPE=S3/AUTO 且结果较大时，/check 返回的 result 不再是结果本身，而是对象存储引用（result_type 为 "s3"，含预签名下载地址）。 Manager 默认就会把这种结果下载好再交给你，调用方拿到的始终是真实结果， 而且响应结构不变（state / result_id 仍在）：

# create_and_wait_result 默认自动下载外置结果
result = m.create_and_wait_result(task_name, params)

# check 默认同样自动下载，并把结果回填到 resp["result"]
resp = m.check(task_name, result_id)          # 完整响应；外置时 result 已是真实结果
resp["state"]                                  # 状态字段照常可用（PENDING/SUCCESS/...）
resp["result"]                                 # 真实结果（不是引用）

# 想拿服务端原始响应（自己用预签名地址下载、查看 sha256/过期时间）
resp = m.check(task_name, result_id, resolve_result=False)   # result 为引用对象

# 也可以显式下载（传响应或直接传预签名地址）
result = m.download_result(resp)                    # 解析 JSON 并返回
m.download_result(resp, local_path="result.json")   # 流式落盘，返回文件路径

异步版本用法相同（await 即可）：

result = await am.create_and_wait_result(task_name, params)
resp = await am.check(task_name, result_id)                        # 默认已下载
resp = await am.check(task_name, result_id, resolve_result=False)   # 原始响应
result = await am.download_result(resp, local_path="result.json")

注：预签名地址自带鉴权，下载时不会携带 Basic 凭据（否则会破坏签名）。

结果形态严格按服务端的 result_type 处理，不做静默兜底：

• json / text：result 内容直接返回（内联结果与错误/状态文本保持原行为）
• s3：取引用里的下载地址并下载解析，把结果回填到 resp["result"]； 引用格式不符、缺下载地址、下载失败均直接报错
• 未知的 result_type、响应不是对象：直接报错（客户端与服务端版本不匹配）
• 服务端未返回 result_type（旧版本）：打印 warning 提醒尽快升级，并按内联结果处理

Upload a file

file_name = m.upload("/path/to/file.txt")

Download a file

m.download("file.txt", "/path/to/save.txt")

Revoke a task

m.revoke(result_id)

Configuration

Architecture

BaseManager (shared initialization and utilities)
├── Manager (synchronous implementation)
│   ├── _req: uses requests library
│   └── _wait: uses time.sleep
└── AsyncManager (asynchronous implementation)
    ├── _req: uses httpx async client
    └── _wait: uses asyncio.sleep