PowerJob Client Python

这是 tech.powerjob:powerjob-client:5.1.2 的 Python 同步客户端移植版，用于调用 PowerJob Server 的 /openApi 接口。它移植的是 OpenAPI 管理客户端，不是接收和执行任务的 Worker SDK。

已覆盖 Java 客户端的 Job、Instance、Workflow、Workflow Instance API，并实现：

• AppName + 密码 MD5 鉴权
• access token 自动刷新
• 多 Server 地址故障切换
• HTTP/HTTPS、自定义请求头和超时
• snake_case 方法及 Java camelCase 兼容别名
• 请求 dataclass、枚举和结构化 Result

安装

python -m pip install .

开发和测试：

python -m pip install -e ".[test]"
pytest

快速开始

from powerjob_client import PowerJobClient

with PowerJobClient("127.0.0.1:7700", "my-app", "powerjob-password") as client:
    jobs = client.fetch_all_job()
    if jobs.success:
        print(jobs.data)

    instance_id = client.run_job(10086, instance_params='{"source":"python"}').unwrap()
    print(instance_id)

集群和 HTTPS 配置：

from powerjob_client import ClientConfig, PowerJobClient

config = ClientConfig(
    app_name="my-app",
    password="secret",
    address_list=["powerjob-1:7700", "powerjob-2:7700"],
    protocol="https",
    connection_timeout=3,
    read_timeout=10,
    default_headers={"X-Tenant": "demo"},
    verify_ssl=True,
    trust_env=False,
)
client = PowerJobClient(config)

为忠实兼容 Java 5.1.2 客户端，HTTPS 的 verify_ssl 默认是 False。生产环境建议显式设为 True 并使用有效证书。

客户端默认不读取 HTTP_PROXY、HTTPS_PROXY、ALL_PROXY 等环境配置，避免访问内网 PowerJob Server 时意外走本机代理。如确实需要继承环境代理，请配置 trust_env=True。

新建或更新 Job

from powerjob_client import (
    ExecuteType,
    PowerJobClient,
    ProcessorType,
    SaveJobInfoRequest,
    TimeExpressionType,
)

request = SaveJobInfoRequest(
    job_name="python-created-job",
    job_description="created through PowerJob OpenAPI",
    processor_info="com.example.DemoProcessor",
    execute_type=ExecuteType.STANDALONE,
    processor_type=ProcessorType.BUILT_IN,
    time_expression_type=TimeExpressionType.CRON,
    time_expression="0 0 * * * ?",
)

with PowerJobClient("127.0.0.1:7700", "my-app", "secret") as client:
    job_id = client.save_job(request).unwrap()

也可直接传字典。字典允许 Python snake_case 键，发送时会转换成 PowerJob 的 camelCase：

client.query_job({"job_name_like": "demo", "status_in": [1]})

API 对照

所有 Java 风格方法名也可直接调用。响应统一为 Result(success, data, message, code)；data 保留 Server 原始 JSON 对象，避免版本升级时因新增字段反序列化失败。

与 Java 版的有意差异

• Python 不区分 Java 的 ResultDTO 和 PowerResultDTO，统一为 Result，并保留可选 code。
• Python 字典响应保持原始形态，不强制转换为大量 DTO 类。
• Java 的 writeTimeout 在 requests 中没有完全对应的独立设置；连接与读取超时分别由 connection_timeout、read_timeout 控制。
• 动态地址发现使用无参 address_provider: Callable[[], Sequence[str]]。

版本范围

实现以 PowerJob 官方仓库标签 v5.1.2 中的 powerjob-client 和 powerjob-common 为依据。若 Server 版本较旧，runJob2、分页查询或鉴权响应头可能不可用。