py-wisharetec-toolkit

Wisharetec SaaS 系统的 Python 客户端库，提供同步和异步接口调用能力。

功能特性

• 封装 SaaS 系统的认证机制，包括登录、令牌刷新和令牌缓存
• 提供同步和异步两种 HTTP 请求模式
• 集成 JSON Schema 校验和 JSONPath 查询工具
• 支持 diskcache 和 Redis 两种缓存后端

安装

使用 uv（推荐）

uv add py-wisharetec-toolkit

使用 pip

pip install py-wisharetec-toolkit

依赖包

httpx>=0.27.0
pydantic>=2.0
jsonpath-ng>=1.5.3
jsonschema>=4.21.0
diskcache>=5.6.3
redis>=4.6.0
py-httpx-toolkit>=1.0.0

项目结构

src/
└── py_wisharetec_toolkit/
    ├── __init__.py          # 包入口模块
    └── saas/
        ├── __init__.py      # SaaS 模块入口，包含 Saas 客户端类
        ├── models.py        # 数据模型（查询条件等）
        ├── responses.py     # 响应模型（分页器等）
        └── utils.py         # 工具函数（JSON Schema 校验、响应模型构建等）

API 说明

Saas 类

初始化

from py_wisharetec_toolkit.saas import Saas

saas = Saas(
    base_url="https://saas.wisharetec.com/",
    account="your_account",
    password="your_password",
    cache_config={
        "instance": None,  # diskcache.Cache 或 redis.Redis 实例
        "key": "py_wisharetec_saas_account",
        "expire": 7100,
    },
    client_kwargs={
        "timeout": 60,
        "verify": False,
    }
)

同步方法

异步方法

utils 工具函数

json_is_valid

校验 JSON 数据是否符合指定的 JSON Schema。

from py_wisharetec_toolkit.saas.utils import json_is_valid

schema = {
    "type": "object",
    "properties": {"id": {"type": "string"}},
    "required": ["id"]
}
data = {"id": "123"}
result = json_is_valid(schema, data)
# 返回: True

json_find_first

使用 JSONPath 表达式从 JSON 数据中查找第一个匹配项。

from py_wisharetec_toolkit.saas.utils import json_find_first

data = {"userInfoV": {"id": "123"}}
result = json_find_first("$.userInfoV.id", data)
# 返回: "123"

json_find

使用 JSONPath 表达式从 JSON 数据中查找所有匹配项。

from py_wisharetec_toolkit.saas.utils import json_find

data = {"items": [{"id": "1"}, {"id": "2"}]}
result = json_find("$.items[*].id", data)
# 返回: [1, 2]

json_find_values

使用 JSONPath 表达式从 JSON 数据中提取所有匹配的值。

json_find_first_value

使用 JSONPath 表达式从 JSON 数据中提取第一个匹配的值。

build_paginator_instance

将 HTTP 响应或字典转换为 Paginator 响应模型。

from py_wisharetec_toolkit.saas.utils import build_paginator_instance

result = build_paginator_instance({
    "pageNum": 1, 
    "pageSize": 50, 
    "total": 100, 
    "last": False, 
    "records": []
})
print(result.records)  # []

build_base_instance

将 HTTP 响应或字典转换为 Base 响应模型。

models 数据模型

QueryCondition

查询条件模型，用于封装分页查询参数。

from py_wisharetec_toolkit.saas.models import QueryCondition

condition = QueryCondition(
    conditions={"name": "test"},
    pageNum=1,
    pageSize=50,
    orderBy=["-createTime"]
)

QueryConditionField

查询条件字段模型，用于定义单个查询条件。

from py_wisharetec_toolkit.saas.models import QueryConditionField

field = QueryConditionField(
    method=1,
    name="status",
    value="active"
)

responses 响应模型

Paginator

分页响应模型，继承自 py_httpx_toolkit.responses.Base。

from py_wisharetec_toolkit.saas.responses import Paginator

paginator = Paginator(
    pageNum=1,
    pageSize=50,
    total=100,
    last=False,
    records=[]
)

使用示例

同步方式

from py_wisharetec_toolkit.saas import Saas

saas = Saas(
    account="test",
    password="password"
)

# 登录获取 Token
saas.login()

# 发送请求
response = saas.request_with_token(
    method="GET",
    url="/api/data/list"
)
print(response.json())

异步方式

import asyncio
from py_wisharetec_toolkit.saas import Saas

async def main():
    saas = Saas(
        account="test",
        password="password"
    )
    
    # 异步登录
    await saas.async_login()
    
    # 异步发送请求
    response = await saas.async_request_with_token(
        method="GET",
        url="/api/data/list"
    )
    print(response.json())

asyncio.run(main())

使用缓存

import diskcache
from py_wisharetec_toolkit.saas import Saas

cache = diskcache.Cache("./cache")

saas = Saas(
    account="test",
    password="password",
    cache_config={
        "instance": cache,
        "key": "saas_token",
        "expire": 7100
    }
)

# 自动从缓存获取或刷新 Token
saas.refresh_token()

作者

Guolei

• 邮箱: 174000902@qq.com

项目主页

• Gitee: https://gitee.com/guolei19850528/py_wisharetec_toolkit

许可证

MIT License