Metadata-Version: 2.4
Name: qdata-adapter-base
Version: 1.0.0
Summary: QDataV2 Application Adapter Framework — async-first base library for building external application adapters
Project-URL: Homepage, https://www.qeasy.cloud
Project-URL: Documentation, https://github.com/qeasy/qdata-adapter-base#readme
Project-URL: Repository, https://github.com/qeasy/qdata-adapter-base
Project-URL: Bug Tracker, https://github.com/qeasy/qdata-adapter-base/issues
Project-URL: Changelog, https://github.com/qeasy/qdata-adapter-base/blob/main/CHANGELOG.md
Author-email: 广东轻亿云软件科技有限公司 <opensource@qeasy.cloud>
Maintainer-email: QData Team <opensource@qeasy.cloud>
License: MIT
License-File: LICENSE
Keywords: adapter,api,async,crm,erp,integration,kingdee,oauth2,qdata,qeasy
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: tenacity>=8.0.0
Provides-Extra: dev
Requires-Dist: mypy>=1.10.0; extra == 'dev'
Requires-Dist: pre-commit>=3.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.0.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.30.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Description-Content-Type: text/markdown

# QData Adapter Base

<p align="center">
  <strong>异步优先的外部应用适配器开发框架</strong>
</p>

<p align="center">
  由 <a href="https://www.qeasy.cloud">广东轻亿云软件科技有限公司</a> 开发<br>
  「轻易云数据集成平台」核心组件
</p>

<p align="center">
  <a href="https://www.python.org/downloads/"><img src="https://img.shields.io/badge/python-3.11+-blue.svg" alt="Python 3.11+"></a>
  <a href="https://opensource.org/licenses/MIT"><img src="https://img.shields.io/badge/License-MIT-yellow.svg" alt="License: MIT"></a>
  <a href="https://pypi.org/project/qdata-adapter-base/"><img src="https://img.shields.io/pypi/v/qdata-adapter-base.svg" alt="PyPI version"></a>
  <a href="https://pypi.org/project/qdata-adapter-base/"><img src="https://img.shields.io/pypi/dm/qdata-adapter-base.svg" alt="Downloads"></a>
  <a href="https://github.com/qeasy/qdata-adapter-base/actions/workflows/ci.yml"><img src="https://github.com/qeasy/qdata-adapter-base/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
  <a href="https://codecov.io/gh/qeasy/qdata-adapter-base"><img src="https://codecov.io/gh/qeasy/qdata-adapter-base/branch/main/graph/badge.svg" alt="Coverage"></a>
</p>

---

## 关于轻易云数据集成平台

> **数据集成，简单看得见**

轻易云是一款企业级数据集成平台，专注于帮助企业快速、高效地打通各类业务系统之间的数据通道。

<a href="https://www.qeasy.cloud">
  <img src="https://qcdn.qeasy.cloud/static/logo.svg" alt="轻易云" height="60">
</a>

| 亮点 | 描述 |
|------|------|
| **即插即用** | 无需复杂开发，配置即连接，支持 500+ 主流应用系统对接 |
| **全程可视** | 数据流动、转换过程、执行状态一目了然 |
| **高性能引擎** | 基于优化的异步执行引擎，轻松处理海量数据同步 |
| **企业级可靠** | 完善的认证管理、重试机制、错误追踪 |

**QData Adapter Base** 是轻易云数据集成平台的适配器开发框架，为对接金蝶、用友、吉客云等 ERP/CRM/WMS 系统提供标准化的开发基座。

---

## 特性

- **异步优先** — 基于 `httpx` 的高性能异步 HTTP 客户端
- **多认证方式** — OAuth2、HMAC 签名、API Key、Session 登录
- **自动重试** — 基于 `tenacity` 的指数退避重试，5xx / 网络错误自动重试
- **组合器模式** — 一个平台多套接口体系（如标准 API + 奇门 API）
- **插件架构** — Entry Points 动态发现适配器，即装即用
- **类型安全** — 完整类型标注 + Pydantic 数据验证 + py.typed
- **连接测试** — 结构化的 `TestConnectionResult` 连接检测

---

## 安装

```bash
pip install qdata-adapter-base
```

开发模式：

```bash
pip install qdata-adapter-base[dev]
```

源码安装：

```bash
git clone https://github.com/qeasy/qdata-adapter-base.git
cd qdata-adapter-base
pip install -e ".[dev]"
```

---

## 快速开始

### 创建适配器

```python
from qdata_adapter import BaseAppAdapter, ConnectorContext, TestConnectionResult
from typing import AsyncIterator, Any
import time


class MyERPAdapter(BaseAppAdapter):
    app_code = "my_erp"
    adapter_version = "1.0.0"

    async def authenticate(self) -> dict[str, Any]:
        response = await self.http_client.post(
            "/api/auth/login",
            json={
                "username": self.context.auth_config["username"],
                "password": self.context.auth_config["password"],
            },
        )
        return {"access_token": response["token"], "expires_in": 3600}

    async def refresh_token(self) -> dict[str, Any]:
        return await self.authenticate()

    async def list_objects(
        self, object_type: str, filters: dict | None = None, page_size: int = 100
    ) -> AsyncIterator[dict[str, Any]]:
        await self.ensure_authenticated()
        page = 1
        while True:
            response = await self.http_client.get(
                f"/api/{object_type}",
                params={"page": page, "page_size": page_size, **(filters or {})},
            )
            for item in response.get("data", []):
                yield item
            if len(response.get("data", [])) < page_size:
                break
            page += 1

    async def get_object(self, object_type: str, object_id: str) -> dict[str, Any]:
        await self.ensure_authenticated()
        return await self.http_client.get(f"/api/{object_type}/{object_id}")

    async def test_connection(self) -> TestConnectionResult:
        start = time.time()
        try:
            await self.authenticate()
            return TestConnectionResult.connected(
                duration_ms=int((time.time() - start) * 1000),
            )
        except Exception as e:
            return TestConnectionResult.network_error(str(e))
```

### 使用适配器

```python
import asyncio
from qdata_adapter import ConnectorContext, AdapterRegistry

# 注册适配器
AdapterRegistry.register(MyERPAdapter)

async def main():
    context = ConnectorContext(
        connector_id="my-connector-001",
        app_software_code="my_erp",
        base_url="https://api.my-erp.com",
        auth_config={"username": "admin", "password": "secret"},
    )

    adapter = AdapterRegistry.create_adapter("my_erp", context)

    # 测试连接
    result = await adapter.test_connection()
    print(f"连接状态: {result.status}, 耗时: {result.duration_ms}ms")

    # 查询数据
    async for order in adapter.list_objects("orders", {"status": "pending"}):
        print(f"Order: {order['id']}")

asyncio.run(main())
```

---

## 核心组件

| 组件 | 说明 |
|------|------|
| `ConnectorContext` | 连接器上下文 — Pydantic 数据验证，敏感信息脱敏 |
| `BaseAppAdapter` | 适配器抽象基类 — 定义标准接口，管理认证生命周期 |
| `AdapterRegistry` | 适配器注册中心 — 手动注册 + Entry Points 自动发现 |
| `HttpClient` | 异步 HTTP 客户端 — 重试、超时、错误转换 |
| `TestConnectionResult` | 连接测试结果 — 结构化的健康检查响应 |
| `TokenCacheProtocol` | Token 缓存协议 — 可插拔（内存 / Redis / 自定义） |

---

## 异常体系

```
AdapterError
├── AuthenticationError
│   └── TokenExpiredError
├── AdapterConnectionError
│   └── AdapterTimeoutError
├── ResponseError
│   ├── RateLimitError
│   └── NotFoundError
├── ValidationError
├── ConfigurationError
└── NodeError
    ├── NodeConfigError
    ├── NodeExecutionError
    ├── NodeInputError / NodeOutputError
    └── ConnectorNotFoundError / ConnectorInactiveError
```

---

## 组合器模式

当一个平台有多套接口体系时（如吉客云的标准接口 + 奇门接口），使用组合器模式：

```
JkyAdapter（唯一外部入口）
├── settings["interface"] = "standard" → JkyStandardInterface
└── settings["interface"] = "qimen"   → JkyQimenInterface
```

适配器包通过 Entry Points 注册，即装即用：

```toml
# pyproject.toml
[project.entry-points."qdata.adapters"]
jky = "qdata_adapter_jky:JkyAdapter"
kingdee = "qdata_adapter_kingdee:KingdeeAdapter"
```

---

## 开发

```bash
git clone https://github.com/qeasy/qdata-adapter-base.git
cd qdata-adapter-base

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

make test          # 运行测试
make lint          # 代码检查
make format        # 代码格式化
make check         # lint + 类型检查
make test-cov      # 测试 + 覆盖率
make build         # 构建 wheel
```

---

## 许可证

本项目采用 **MIT License** — 详见 [LICENSE](LICENSE)。

---

## 关于我们

**广东轻亿云软件科技有限公司**
专注数据集成与处理，提供企业级 ETL/ELT 解决方案

| | |
|---|---|
| 官网 | [https://www.qeasy.cloud](https://www.qeasy.cloud) |
| 开源项目 | opensource@qeasy.cloud |
| 商业咨询 | vincent@qeasy.cloud |

---

*Powered by [广东轻亿云软件科技有限公司](https://www.qeasy.cloud)*
