qingflow-sdk 1.0.6

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

轻流 OpenAPI Python SDK

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

✨ 特性

• 功能完整：覆盖 40+ API 接口，涵盖应用管理、环境管理、通讯录管理
• AppDataModel 字段映射：声明式定义 que_id → 属性映射，一行代码完成 API 响应解析
• Pythonic 设计：类型提示、异步支持、上下文管理
• 同步/异步双模式：基于 httpx，一套代码同时支持 sync 和 async
• 自动重试：指数退避重试策略，应对网络波动
• 自动分页：内置翻页功能，轻松获取全量数据
• 精确查询：searchKeys 精准筛选，避免前缀匹配歧义
• 完善异常体系：细粒度异常分类，便于错误处理
• 生产就绪：连接池管理、超时控制、结构化日志

📦 安装

# 从 PyPI 安装
pip install qingflow-sdk

# 或从本地安装
pip install /path/to/qingflow-sdk-python/

🚀 快速开始

1. 声明式字段映射（推荐）

通过 AppDataModel + app_data_field 定义字段映射，一行代码完成 API 响应解析：

from qingflow import QingflowClient, GetAppDataRequest, QueryCondition
from qingflow.models.app_data import AppDataModel, app_data_field

# 定义数据模型 — 声明 que_id 与属性的映射关系
class TestData(AppDataModel):
    app_key: str = "ec12g2201002"
    
    name: str = app_data_field(que_id=345614786, que_title="姓名")
    age: int = app_data_field(que_id=345614787, que_title="年龄", default=0)
    birthday: str = app_data_field(que_id=345614788, que_title="生日")
    gender: str = app_data_field(que_id=345614789, que_title="性别")
    status: str = app_data_field(que_id=4, que_title="当前流程状态")

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

# 查询全部数据
request = GetAppDataRequest(page_num=1, page_size=20)
response = client.app.get_app_data("ec12g2201002", request)

# 一行代码完成解析！
records = TestData.from_answers(response.result.result)

for r in records:
    print(f"姓名={r.name}, 年龄={r.age}, 生日={r.birthday}")
    print(f"apply_id={r.apply_id}")

2. 精确条件查询

使用 QueryCondition.exact_match() 进行精准筛选（内部使用 searchKeys 参数）：

# 精确查询：姓名 = "快乐小马"
request = GetAppDataRequest(
    page_num=1,
    page_size=10,
    queries=[QueryCondition.exact_match(que_id=345614786, value="快乐小马")]
)
response = client.app.get_app_data("ec12g2201002", request)
records = TestData.from_answers(response.result.result)

# searchKey（前缀匹配）vs searchKeys（精确匹配）对比：
# searchKey='快乐小马'     → 匹配 2 条（含"快乐小马马"）
# searchKeys=['快乐小马']  → 匹配 1 条（仅"快乐小马"）

3. 多条件组合查询

request = GetAppDataRequest(
    page_num=1,
    page_size=10,
    queries=[
        QueryCondition.exact_match(que_id=345614789, value="女"),  # 性别=女
    ]
)
response = client.app.get_app_data("ec12g2201002", request)
records = TestData.from_answers(response.result.result)

4. JSON 输出

import json

records = TestData.from_answers(response.result.result)
json_data = [r.to_dict() for r in records]
print(json.dumps(json_data, ensure_ascii=False, indent=2))
# 输出：
# [
#   {
#     "姓名": "快乐小马",
#     "年龄": 28,
#     "生日": "2026-03-04",
#     "性别": "女",
#     "当前流程状态": "已通过"
#   }
# ]

5. 模型 → 创建数据请求

from qingflow import AddAppDataRequest

# 创建模型实例
new_record = TestData(
    name="测试用户",
    age=25,
    birthday="2026-05-19",
    gender="男",
)

# 转换为 AnswerDTO 列表
answers = new_record.to_answers()

# 构建创建请求
request = AddAppDataRequest(answers=answers)
response = client.app.add_app_data("ec12g2201002", request)
print(f"创建成功，Apply ID: {response.result.apply_id}")

6. 基础使用（不使用 AppDataModel）

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}")

7. 添加数据

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}")

8. 异步使用

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())

9. 自动分页获取全量数据

# 自动翻页获取所有用户
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)

10. 上下文管理

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

📚 模块说明

应用管理（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=***
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 参考实现。