Metadata-Version: 2.4
Name: qingflow-sdk
Version: 1.0.6
Summary: 轻流 OpenAPI Python SDK - 基于 httpx 的同步/异步客户端
Author-email: 高文伟 <gaowenwei97@126.com>
License: MIT
Project-URL: Homepage, https://github.com/gwwgogogo/qingflow-sdk-python
Project-URL: Documentation, https://qingflow-sdk.readthedocs.io
Project-URL: Repository, https://github.com/gwwgogogo/qingflow-sdk-python
Keywords: qingflow,sdk,api,low-code
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: pydantic-settings>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: ruff>=0.4.0; extra == "dev"
Requires-Dist: mypy>=1.0; extra == "dev"
Provides-Extra: docs
Requires-Dist: mkdocs>=1.6; extra == "docs"
Requires-Dist: mkdocstrings[python]>=0.25; extra == "docs"
Dynamic: license-file

# 轻流 OpenAPI Python SDK

[![Python Version](https://img.shields.io/badge/python-3.9+-blue.svg)](https://www.python.org/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![PyPI version](https://img.shields.io/badge/pypi-v1.0.5-blue.svg)](https://pypi.org/project/qingflow-sdk/)

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

## ✨ 特性

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

## 📦 安装

```bash
# 从 PyPI 安装
pip install qingflow-sdk

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

## 🚀 快速开始

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

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

```python
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` 参数）：

```python
# 精确查询：姓名 = "快乐小马"
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. 多条件组合查询

```python
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 输出

```python
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. 模型 → 创建数据请求

```python
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）

```python
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. 添加数据

```python
from qingflow import AddAppDataRequest, AnswerDTO, ValueDTO

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

### 8. 异步使用

```python
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. 自动分页获取全量数据

```python
# 自动翻页获取所有用户
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. 上下文管理

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

## 📚 模块说明

### 应用管理（App）

```python
# 获取应用数据
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）

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

```python
# 获取环境级 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）

```python
# 获取用户列表
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")
```

## ⚙️ 配置说明

### 环境变量

```bash
# .env 文件
QINGFLOW_ACCESS_TOKEN=***
QINGFLOW_BASE_DOMAIN=https://api.qingflow.com
QINGFLOW_TIMEOUT=30.0
QINGFLOW_MAX_RETRIES=3
QINGFLOW_LOG_LEVEL=INFO
```

### 代码配置

```python
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,
)
```

## 🧪 开发测试

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

# 运行测试
pytest

# 代码检查
ruff check .
mypy .
```

## 📄 许可证

MIT License - 详见 [LICENSE](LICENSE) 文件

## 👤 作者

**高文伟**

## 🙏 致谢

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