py-tjbrhk-toolkit

天津博瑞皓科 Python 客户端库，提供与天津博瑞皓科 API 交互的完整功能支持。

特性

• 支持同步和异步两种请求模式
• 提供通知发送功能
• 基于 Pydantic 的数据验证
• JSONPath 数据查询支持
• JSON Schema 校验功能

安装

使用 uv（推荐）

uv add py-tjbrhk-toolkit

使用 pip

pip install py-tjbrhk-toolkit

依赖包

依赖	版本要求	说明
httpx	>=0.27.0	HTTP 客户端库
pydantic	>=2.0	数据验证库
jsonpath-ng	>=1.5.3	JSONPath 查询库
jsonschema	>=4.21.0	JSON Schema 校验库
py-httpx-toolkit	>=1.0.1	HTTP 工具库

快速开始

同步模式

from py_tjbrhk_toolkit.speaker import Speaker

# 初始化客户端
client = Speaker(
    base_url="https://speaker.17laimai.cn",
    token="your_token",
    id="your_id",
    version="1"
)

# 发送通知
response = client.notify(message="Hello World")
print(response.json())

异步模式

import asyncio
from py_tjbrhk_toolkit.speaker import Speaker

async def main():
    # 初始化客户端
    client = Speaker(
        base_url="https://speaker.17laimai.cn",
        token="your_token",
        id="your_id",
        version="1"
    )
    
    # 异步发送通知
    response = await client.async_notify(message="Hello World")
    print(response.json())

asyncio.run(main())

API 文档

Speaker 客户端

初始化参数

参数	类型	默认值	说明
base_url	str	https://speaker.17laimai.cn	Speaker 服务器地址
token	str	""	API 认证 Token
id	str	""	API 认证 ID
version	str	"1"	API 版本号
client_kwargs	dict	None	httpx.Client 额外配置

notify(message, client=None, client_kwargs=None, **kwargs)

发送同步通知请求

参数:

• message: 要发送的内容
• client: 自定义 httpx.Client 实例
• client_kwargs: 客户端配置参数
• **kwargs: 传递给 httpx 请求的额外参数

返回: httpx.Response - HTTP 响应对象

async_notify(message, client=None, client_kwargs=None, **kwargs)

发送异步通知请求

参数:

• message: 要发送的内容
• client: 自定义 httpx.AsyncClient 实例
• client_kwargs: 客户端配置参数
• **kwargs: 传递给 httpx 请求的额外参数

返回: httpx.Response - HTTP 响应对象

工具函数

json_find_first(expression, data)

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

from py_tjbrhk_toolkit.speaker.utils import json_find_first

data = {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}
result = json_find_first("$.users[0].name", data)
print(result)  # Alice

json_is_valid(schema, data)

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

from py_tjbrhk_toolkit.speaker.utils import json_is_valid

schema = {"type": "object", "properties": {"name": {"type": "string"}}}
data = {"name": "test"}
valid = json_is_valid(schema, data)
print(valid)  # True

build_success_instance(response)

将 HTTP 响应或字典转换为 Success 模型对象

from py_tjbrhk_toolkit.speaker.utils import build_success_instance

response = {"errcode": 0, "errmsg": "success"}
result = build_success_instance(response)
print(result.errcode)  # 0

success_is_valid(response, schema=None)

校验 HTTP 响应或字典是否符合成功响应的 JSON Schema

from py_tjbrhk_toolkit.speaker.utils import success_is_valid

response = {"errcode": 0, "errmsg": "success"}
valid = success_is_valid(response)
print(valid)  # True

响应模型

Base

响应基类，包含错误码和错误信息字段

from py_tjbrhk_toolkit.speaker.responses import Base

response = Base(errcode=0, errmsg="success")

Success

成功响应模型，继承自 Base

from py_tjbrhk_toolkit.speaker.responses import Success

response = Success(errcode=0, errmsg="success")

示例代码

完整示例

from py_tjbrhk_toolkit.speaker import Speaker
from py_tjbrhk_toolkit.speaker.utils import build_success_instance, success_is_valid

# 初始化客户端
client = Speaker(
    token="your_token",
    id="your_id"
)

# 发送通知
response = client.notify(message="测试通知")

# 验证响应
if success_is_valid(response):
    result = build_success_instance(response)
    print(f"发送成功: {result.errmsg}")
else:
    print("发送失败")

异步完整示例

import asyncio
from py_tjbrhk_toolkit.speaker import Speaker
from py_tjbrhk_toolkit.speaker.utils import build_success_instance, success_is_valid

async def send_notification():
    client = Speaker(
        token="your_token",
        id="your_id"
    )
    
    response = await client.async_notify(message="异步测试通知")
    
    if success_is_valid(response):
        result = build_success_instance(response)
        print(f"发送成功: {result.errmsg}")
    else:
        print("发送失败")

asyncio.run(send_notification())

作者

Guolei 174000902@qq.com

项目主页

https://gitee.com/guolei19850528/py_tjbrhk_toolkit

官方文档

https://www.yuque.com/lingdutuandui/ugcpag/umbzsd

许可证

MIT License