fasttask_manager
Manager for fasttask
Installation
pip install fasttask_manager
Usage
Create a Manager
Synchronous Manager:
from fasttask_manager import Manager
m = Manager("127.0.0.1", port=8080, protocol="https")
Asynchronous Manager:
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
| Parameter | Type | Default | Description |
|---|---|---|---|
| host | str | - | Server host |
| protocol | str | "http" | HTTP protocol |
| port | int | 80 | Server port |
| tries | int | 5 | Max retry attempts |
| delay | int | 3 | Retry delay (seconds) |
| logger | Logger | None | Custom logger |
| log_prefix | str | "" | Log prefix |
| auth_user | str | "" | Basic auth username |
| auth_passwd | str | "" | Basic auth password |
| url_base_path | str | "" | URL base path |
| req_timeout | int | 30 | Request timeout (seconds) |
| simple_error_log | bool | True | Simple error logging |
| verify_ssl | bool | False | Verify SSL certificate |
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
1 maintainer
Metadata
1 maintainer
Release files for fasttask-manager 0.7.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| fasttask_manager-0.7.0.tar.gz | 14.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| fasttask_manager-0.7.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 31.6 kB
Release files / fasttask_manager-0.7.0.tar.gz
| Download URL | fasttask_manager-0.7.0.tar.gz |
|---|---|
| Size | 14.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
c3bffa372852dec81a7dd8672d22f8ec63b1d05527785f7e723ef74be997acfe
|
|
BLAKE2b-256 checksum How to use checksums |
7996935510abfc65900b2ed828ec16272596a1bea010e9535d8093e87f7e66e7
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.14.6
|
Release files / fasttask_manager-0.7.0-py3-none-any.whl
| Download URL | fasttask_manager-0.7.0-py3-none-any.whl |
|---|---|
| Size | 17.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
b678d29f6807ed94e9e837ce3af5788de7678df9f576bb943c0dc52e2339e858
|
|
BLAKE2b-256 checksum How to use checksums |
f992ee956f5696352f1b363bc4a638728be5f83982d54328147f50d544ba9459
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.14.6
|
