Metadata-Version: 2.4
Name: yz-dubbo
Version: 0.1.0
Summary: 有赞 Dubbo SDK - 支持 Dubbo RPC 调用
Author-email: YouzanTech <tech@youzan.com>
License: MIT
Project-URL: Homepage, https://github.com/youzan/yz-dubbo
Project-URL: Documentation, https://github.com/youzan/yz-dubbo/blob/main/README.md
Project-URL: Repository, https://github.com/youzan/yz-dubbo.git
Project-URL: Issues, https://github.com/youzan/yz-dubbo/issues
Keywords: dubbo,rpc,youzan
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.24.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: flake8>=6.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"

# YZ-Dubbo

有赞 Dubbo SDK - 基于 Tether 网关的 Dubbo RPC 调用封装

## 功能特性

- 🚀 **简单易用**: 提供简洁的函数式 API
- 🌍 **多环境支持**: 内置 dev/test/staging/prod 环境配置
- 🔧 **完整的错误处理**: 统一的错误码体系
- ⏱️ **超时控制**: 支持自定义超时时间,默认3秒
- 🔒 **类型安全**: 完整的类型注解支持

## 安装

```bash
# 开发模式安装
cd packages/yz_dubbo
pip install -e .

# 或从 requirements.txt 安装
pip install -r requirements.txt
```

## 快速开始

### 基础调用

```python
from yz_dubbo import invoke

# 调用 Dubbo 服务
result = invoke(
    service_name="com.youzan.service.UserService",
    method_name="getUserInfo",
    args=[{"userId": 123}]
)

print(result)  # 直接返回响应数据
```

### 自定义 Headers

```python
result = invoke(
    service_name="com.youzan.service.OrderService",
    method_name="createOrder",
    args=[{"productId": 456, "quantity": 1}],
    headers={
        "X-Request-Id": "req-123",
        "X-Tenant-Id": "tenant-456"
    }
)
```

### 设置超时时间

```python
result = invoke(
    service_name="com.youzan.service.PaymentService",
    method_name="pay",
    args=[payment_data],
    timeout=10000  # 10秒超时
)
```

## 环境配置

### 内置环境

SDK 内置了以下环境配置:

| 环境 | 配置值 | Tether 地址 |
|------|--------|------------|
| dev | `env="dev"` | tether-dev.youzan.com |
| test | `env="test"` | tether-test.youzan.com |
| staging | `env="staging"` | tether-staging.youzan.com |
| prod | `env="prod"` | tether.youzan.com |

### 环境变量

```bash
# 设置环境
export DUBBO_ENV=prod

# 或直接设置 Tether 地址
export TETHER_DUBBO_HOST=custom-tether.youzan.com

# 设置应用名
export APPLICATION_NAME=my-service
```

### 代码配置

```python
from yz_dubbo import DubboConfig, DubboClient

# 使用预设环境
config = DubboConfig(env="prod")

# 自定义 host
config = DubboConfig(host="custom-tether.youzan.com", port=80)

# 使用自定义配置
client = DubboClient(config=config)
result = client.invoke(
    service_name="com.youzan.service.UserService",
    method_name="getUser",
    args=[{"userId": 123}]
)
client.close()
```

## 错误处理

### 错误码对照表

| 错误码 | 说明 | HTTP Status |
|--------|------|-------------|
| `ERR_CLIENT_ERROR` | 客户端错误 | 400 |
| `ERR_TIMEOUT` | 请求超时 | 504 |
| `ERR_PARAMS_TYPE_ERROR` | 参数类型错误 | 400 |
| `ERR_FLOW_LIMIT` | 流量限制 | 429 |
| `ERR_SERVICE_UNAVAILABLE` | 服务不可用 | 503 |
| `ERR_INTERNAL_SERVER_ERROR` | 内部错误 | 500 |
| `ERR_NETWORK_ERROR` | 网络错误 | 503 |

### 异常处理示例

```python
from yz_dubbo import invoke, DubboError, ErrorCode

try:
    result = invoke(
        service_name="com.youzan.service.UserService",
        method_name="getUser",
        args=[{"userId": 123}]
    )
    print(f"成功: {result}")

except DubboError as e:
    print(f"错误码: {e.code}")
    print(f"错误信息: {e.message}")
    print(f"上下文: {e.context}")

    # 根据错误码处理
    if e.code == ErrorCode.ERR_TIMEOUT:
        print("请求超时,请稍后重试")
    elif e.code == ErrorCode.ERR_NETWORK_ERROR:
        print("网络错误")
    else:
        print("其他错误")
```

## API 参考

### invoke()

```python
def invoke(
    service_name: str,
    method_name: str,
    args: Optional[List[Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: int = 3000,
) -> Any
```

**参数**:
- `service_name`: 服务接口名 (如 `com.youzan.service.UserService`)
- `method_name`: 方法名
- `args`: 参数列表 (默认为 `[]`)
- `headers`: 自定义请求头 (默认为 `{}`)
- `timeout`: 超时时间,单位毫秒 (默认 `3000`)

**返回值**: 直接返回服务响应的 JSON 数据

**异常**: 抛出 `DubboError`

### DubboClient

```python
class DubboClient:
    def __init__(self, config: Optional[DubboConfig] = None) -> None
    def invoke(
        self,
        service_name: str,
        method_name: str,
        args: Optional[List[Any]] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: int = 3000
    ) -> Any
    def close(self) -> None
```

### DubboConfig

```python
class DubboConfig:
    def __init__(
        self,
        env: Optional[str] = None,
        host: Optional[str] = None,
        port: int = 80,
        timeout: int = 3000,
        from_app: Optional[str] = None
    ) -> None
```

## 实际使用示例

基于真实的 Tether 网关调用:

```bash
# 原始 curl 命令
curl -XPOST http://tether-pre.s.qima-inc.com:8680/soa/com.youzan.channels.apps.service.weapp.WeappPrivacyInterfaceService/batchApplyWithDefault \
  -H'x-request-protocol: dubbo' \
  -H'content-type: application/json' \
  -d '[{"kdtId":111979041}]'
```

对应的 Python 代码:

```python
from yz_dubbo import invoke

result = invoke(
    service_name="com.youzan.channels.apps.service.weapp.WeappPrivacyInterfaceService",
    method_name="batchApplyWithDefault",
    args=[{"kdtId": 111979041}]
)
print(result)
```

## 测试

### 运行测试

```bash
# 运行所有测试
pytest tests/yz_dubbo/ -v

# 运行简化版测试
python tests/yz_dubbo/test_simplified.py
```

### Mock 测试示例

```python
from unittest.mock import Mock, patch
from yz_dubbo import invoke

with patch("yz_dubbo.client.httpx.Client") as mock_client_class:
    # Mock 响应
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {"success": True, "userId": 123}

    mock_http_client = Mock()
    mock_http_client.post.return_value = mock_response
    mock_client_class.return_value = mock_http_client

    # 重置客户端
    from yz_dubbo import client as client_module
    client_module.DubboClient._instance = None
    client_module._default_client = client_module.DubboClient()

    # 调用
    result = invoke(
        service_name="com.youzan.service.UserService",
        method_name="getUser",
        args=[{"userId": 123}]
    )

    assert result == {"success": True, "userId": 123}
```

## 项目结构

```
yz_dubbo/
├── __init__.py         # 对外 API 导出
├── client.py           # 核心客户端实现
├── config.py           # 配置管理
├── errors.py           # 错误定义
└── py.typed            # 类型标记
```

## 常见问题

### Q: 如何切换环境?

**A**: 有三种方式:
```python
# 1. 环境变量
export DUBBO_ENV=prod

# 2. 代码配置
from yz_dubbo import DubboConfig
config = DubboConfig(env="prod")

# 3. 直接指定 host
config = DubboConfig(host="tether.youzan.com")
```

### Q: 超时时间如何设置?

**A**: 可以全局设置或单次调用设置:
```python
# 全局设置
config = DubboConfig(timeout=10000)

# 单次调用 (推荐)
invoke(..., timeout=5000)
```

### Q: 如何添加自定义 Headers?

**A**: 通过 `headers` 参数传递:
```python
result = invoke(
    service_name="com.youzan.service.UserService",
    method_name="getUser",
    args=[{"userId": 123}],
    headers={
        "X-Request-Id": "req-123",
        "X-Custom-Header": "custom-value"
    }
)
```

### Q: 返回值是什么格式?

**A**: 直接返回服务响应的 JSON 数据,不做额外包装:
```python
result = invoke(...)
# result 就是服务返回的原始数据,可能是 dict、list 或其他 JSON 类型
```

## 版本历史

### v0.1.0 (2025-12-30)

**初始版本**
- ✅ 基于 Tether HTTP 网关的 Dubbo 调用
- ✅ 多环境配置支持 (dev/test/staging/prod)
- ✅ 完整的错误码体系
- ✅ 自定义 Headers 支持
- ✅ 超时控制 (默认3秒)
- ✅ 完整的单元测试覆盖

## 许可证

MIT License

## 联系方式

- 项目地址: `/packages/yz_dubbo`
- 测试目录: `/tests/yz_dubbo`

---

**YZ-Dubbo - 让 Dubbo 调用更简单** 🚀
