qingflow-sdk 1.0.3

轻流 OpenAPI Python SDK - 基于 httpx 的同步/异步客户端

轻流 OpenAPI Python SDK

轻流无代码平台 OpenAPI 的 Python SDK，基于官方 Java SDK 功能对等封装。

✨ 特性

• 功能完整：覆盖 40+ API 接口，涵盖应用管理、环境管理、通讯录管理
• Pythonic 设计：类型提示、异步支持、上下文管理
• 同步/异步双模式：基于 httpx，一套代码同时支持 sync 和 async
• 自动重试：指数退避重试策略，应对网络波动
• 分页工具：内置自动翻页功能，轻松获取全量数据
• 完善异常体系：细粒度异常分类，便于错误处理
• 生产就绪：连接池管理、超时控制、结构化日志

📦 安装

# 从本地安装
pip install -e .

# 或从 PyPI 安装（待发布）
pip install qingflow-sdk

🚀 快速开始

基础使用

from qingflow import QingflowClient, GetAppDataRequest, QueryCondition

# 初始化客户端
client = QingflowClient(access_token="your-access-token")

# 查询应用数据
request = GetAppDataRequest(
    page_num=1,
    page_size=20,
    queries=[QueryCondition(que_id=1001, search_key="张三")]
)
response = client.app.get_app_data("customer_mgmt", request)

if response.is_success:
    for item in response.result.result:
        print(f"Apply ID: {item.apply_id}")
else:
    print(f"错误：{response.err_msg}")

添加数据

from qingflow import AddAppDataRequest, AnswerDTO, ValueDTO

request = AddAppDataRequest(
    answers=[
        AnswerDTO(
            que_id=1001,
            values=[ValueDTO(value="张三")]
        ),
        AnswerDTO(
            que_id=1002,
            values=[ValueDTO(value="[e-mail]")]
        ),
    ]
)
response = client.app.add_app_data("customer_mgmt", request)
print(f"添加成功，Apply ID: {response.result.apply_id}")

异步使用

import asyncio
from qingflow import QingflowClient

async def main():
    async with QingflowClient(access_token="your-token", client_type="async") as client:
        response = await client.app.async_get_app_data("app_key", request)
        print(response.result)

asyncio.run(main())

自动分页获取全量数据

# 自动翻页获取所有用户
all_users = client.contact.loop_get_all_users(page_size=200)
print(f"共获取 {len(all_users)} 个用户")

# 自动翻页获取所有应用数据
all_data = client.app.loop_get_all_app_data("app_key", page_size=200)

上下文管理

with QingflowClient(access_token="your-token") as client:
    response = client.app.get_app_data("app_key", request)
# 自动关闭连接

临时切换 Token

with client.with_token("temp-token"):
    response = client.app.get_app_data("app_key", request)
# 自动恢复原 token

📚 模块说明

应用管理（App）

# 获取应用数据
client.app.get_app_data(app_key, request)

# 添加数据
client.app.add_app_data(app_key, request)

# 更新数据
client.app.update_app_data(apply_id, request)

# 删除数据
client.app.delete_app_data(app_key, request)

# 插入或更新（根据主键判断）
client.app.insert_or_update_data(app_key, primary_key_que_id, primary_key_value, request)

# 流程审批
client.app.audit(apply_id, user_id, request)

# 获取流程日志
client.app.audit_record(apply_id)

# 获取报表数据
client.app.get_chart_data(chart_key, request)

# 获取应用列表
client.app.get_app_list(request)

环境管理（Env）

注意：以下接口仅限专有云环境使用

# 获取环境级 Token
client.env.get_admin_access_token(environ_id, env_secret)

# 创建用户
client.env.create_qingflow_user(admin_access_token, request)

# 创建工作区
client.env.create_workspace(admin_access_token, request)

# 获取工作区信息
client.env.get_workspace_info_with_ws_id(admin_access_token, ws_id)

通讯录管理（Contact）

# 获取用户列表
client.contact.get_users(page)

# 获取单个用户
client.contact.get_single_user(user_id)

# 创建用户
client.contact.create_user(request)

# 获取部门列表
client.contact.get_departments()

# 获取部门成员
client.contact.get_department_members(dept_id, fetch_child=True)

# 获取角色列表
client.contact.get_roles()

# 根据手机号或邮箱查询用户
client.contact.get_user_by_email_or_mobile(mobile="13800138000")

⚙️ 配置说明

环境变量

# .env 文件
QINGFLOW_ACCESS_TOKEN=your-access-token
QINGFLOW_BASE_DOMAIN=https://api.qingflow.com
QINGFLOW_TIMEOUT=30.0
QINGFLOW_MAX_RETRIES=3
QINGFLOW_LOG_LEVEL=INFO

代码配置

client = QingflowClient(
    access_token="your-token",
    base_domain="https://api.qingflow.com",
    timeout=60.0,
    max_retries=5,
    auto_assert_success=True,
    log_requests=True,
    log_responses=True,
)

🧪 开发测试

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试
pytest

# 代码检查
ruff check .
mypy .

📄 许可证

MIT License - 详见 LICENSE 文件

👤 作者

高文伟

🙏 致谢

本项目基于轻流官方 Java SDK 的功能实现，感谢轻流团队提供的 API 文档和 Java SDK 参考实现。