调用接口

本页介绍用 Python SDK 调用服务端 REST API 的通用规则：REST 接口与 SDK 方法的对应方式、返回值、分页与自动翻页、批量接口的结果、错误的判断、SDK 的重试与限流处理、文件的上传与下载、应用状态的影响、日志与排查，以及在 asyncio 程序和单元测试中的用法。

各接口的参数和语义见服务端 REST API，每个方法对应哪个接口见服务与方法。下面的示例中，client 是创建客户端得到的 DRClient；使用 AsyncDRClient 时在调用前加 await。

命名规则

REST 接口按模块分为 13 个服务，每个服务是客户端的一个属性：client.users、client.friends、client.blacklist、client.groups、client.messages、client.conversations、client.media、client.push、client.chatrooms、client.moderation、client.calls、client.channels、client.presence。

方法名是 snake_case 的动词或动词短语：

REST 接口	SDK 方法
`POST /users` 创建用户	`client.users.create(...)`
`GET /users/{username}` 查询用户	`client.users.get(username)`
`GET /users` 分页列出用户	`client.users.list(...)`，自动翻页为 `client.users.iter(...)`
`GET /groups/{group_id}/members` 列出群成员	`client.groups.list_members(group_id)`，自动翻页为 `iter_members`
`GET /groups/{group_id}/members/{username}` 查询一位成员	`client.groups.get_member(group_id, username)`
`GET /users/{username}/push-settings` 查询推送设置	`client.push.get_settings(username)`
`PUT`、`DELETE /groups/{group_id}/mute-all` 开启、关闭全员禁言	`client.groups.set_mute_all(group_id, True)`、`set_mute_all(group_id, False)`

返回列表的方法叫 list 或 list_xxx，返回单个对象或一组设置的叫 get 或 get_xxx，分页的列表另有遍历方法 iter 或 iter_xxx；统计和汇总类的保留原名，如 conversations.unread_count、media.usage、moderation.stats、chatrooms.online_counts。全部方法见服务与方法。

参数与 REST 接口的字段同名（服务端的字段名本来就是 snake_case）：

位置参数：路径中的标识（用户名、群 ID、消息 ID 等，按路径中的顺序）；请求体只有一个必填字段时的那个字段，如 users.set_password(username, password)、groups.set_role(group_id, username, role)；批量接口的项列表，如 users.batch_create(users)、messages.batch_send(to_users, ...)。例外是 calls.stats(from_, to) 的两个日期。
关键字参数：其余参数一律必须写参数名，如 users.create("zhangsan", nickname="张三")。这样以后接口增加参数时，不会悄悄传错位置。
与 Python 关键字同名的参数加下划线：发送者 from 写作 from_，请求中仍是 from。type、id 等与内置函数同名的照常使用。
ID 一律是 str：传 int 的在本地以 invalid_params 拒绝，不会发出请求（服务端不接受数字形式的 ID）。seq、版本号是 int。
时间：参数可以传带时区的 datetime（转为 UTC 的 RFC 3339 字符串）或字符串；不带时区的 datetime 在本地以 invalid_params 拒绝，SDK 不猜测时区。响应中的时间是带 UTC 时区的 datetime。

不传、传 null 与传值

修改类方法的可选参数默认值是 UNSET，表示不出现在请求中；传 None 表示发送 null；传其他值照常发送：

python

from deeprespond_im import DRClient


def update_profile(client: DRClient) -> None:
    # 请求体：{"nickname":"张三","attributes":{"vip":"1","level":null}}
    client.users.update("zhangsan", nickname="张三", attributes={"vip": "1", "level": None})

    # attributes=None 发送 "attributes": null，清空全部自定义属性
    client.users.update("zhangsan", attributes=None)

    # friend_add_mode 是必须出现、可以为 null 的字段：None 表示改用应用的默认值
    client.friends.set_settings("zhangsan", None)

参数的类型标注区分了这三种情况（如 str | None | Unset），类型检查器能发现把 None 传给不接受 null 的参数；运行时收到的同样在本地以 invalid_params 拒绝。

原样发送 JSON

消息的 body、ext 等自由内容可以传 dict，SDK 按紧凑的 UTF-8 JSON 编码（不把中文转义为 \uXXXX，拒绝 NaN 和 Infinity）。需要保留数字的写法（如 1.00）或你对内容做了签名时，用 RawJSON 原样发送，SDK 只检查它是一个合法的 JSON 对象：

python

from deeprespond_im import DRClient, RawJSON


def send_price_card(client: DRClient) -> None:
    client.messages.send(
        from_="notice",
        to_user="zhangsan",
        type="custom",
        body=RawJSON('{"card":"price","amount":1.00}'),
    )

RawJSON 只能用在自由内容的位置，放进普通参数里的以 invalid_params 拒绝。服务端保存消息时会去掉 JSON 中的空白，对内容签名的请基于紧凑写法。

每个方法的通用参数

除了接口自己的参数，每个方法都可以传以下三个参数：

参数	说明
`timeout`	这次调用中每次尝试的时限（秒），覆盖客户端的 `timeout`
`deadline`	这次调用的总时限（秒），包括全部重试、限流等待和换取 App Token 的时间。到时抛出 `DRTimeoutError`，不再重试
`request_id`	追踪 ID，作为请求头 `X-Request-ID` 发送，服务端的日志和错误响应中使用同一个 ID。1 到 64 个字母、数字或 `-`、`_`、`.`，不给时 SDK 生成

python

from deeprespond_im import DRClient


def get_user_for_web(client: DRClient, username: str, trace_id: str) -> str:
    # 在线请求中最多花 2 秒，日志用业务的追踪 ID 关联
    user = client.users.get(username, deadline=2.0, request_id=trace_id)
    return user.nickname

遍历方法（iter、iter_xxx）只有 timeout 和 request_id，没有 deadline。

返回值

响应解析为 deeprespond_im.types 中的数据类，字段名与 REST 文档相同：

python

from deeprespond_im import DRClient


def show_user(client: DRClient) -> None:
    user = client.users.get("zhangsan")
    print(user.username, user.nickname, user.status, user.created_at)   # created_at 是带时区的 datetime
    print(user.raw.get("some_new_field"))   # raw：这个对象在响应中的原始内容
    data = user.to_dict()                   # 原始内容的副本，可以直接作为 FastAPI 等的返回值
    print(data["username"])

数据类不可修改，== 按业务字段比较；不可哈希，不能作为 dict 的键或放进 set；
新增的字段：服务端以后增加的字段不必等 SDK 升级，从 raw 中读取。枚举类的字段是 str，遇到新的取值也不会报错；
消息的原文：含消息的对象（Message、SendResult、回调事件等）除了解析后的 body、ext（dict），另有 body_raw、ext_raw（RawJSON），是响应中这一段的原始文本，与服务端返回的逐字节相同，可以直接用于验签或转发；
repr() 不显示 raw、*_raw 和登录凭证、下载地址等敏感字段，把对象写进日志不会泄露它们；
REST 接口返回 204 No Content 的方法返回 None。

分页

分页的列表方法返回一页结果。按游标分页的是 Page，next_cursor 为 None 表示没有下一页：

python

from deeprespond_im import DRClient


def all_active_usernames(client: DRClient) -> list[str]:
    names: list[str] = []
    page = client.users.list(status="active", limit=100)
    while True:
        names.extend(user.username for user in page.items)
        if page.next_cursor is None:
            return names
        page = client.users.list(status="active", limit=100, cursor=page.next_cursor)

类型	用于	字段
`Page[T]`	大多数列表	`items`、`next_cursor`、`raw`
`VersionedPage[T]`	好友列表、黑名单、群成员等带版本号的列表	`Page` 的字段，加 `version`、`not_modified`、`count`（好友数、黑名单人数、群成员数）、`max_count`（好友、黑名单的上限）；没有的为 `None`
`SeqPage[Message]`	会话消息 `conversations.list_messages`	`items`（按序号从小到大）、`has_more`、`next_after_seq`、`next_before_seq`，按 `around_seq` 取时另有 `has_more_before`、`has_more_after`
`ExportPage[Message]`	消息导出 `messages.export`	`items`、`has_more`、`next_after_message_id`

带版本号的列表可以传上次得到的 version 作为 known_version，没有变化时 not_modified 为 True、items 为空，不必重新拉取：

python

from deeprespond_im import DRClient


def friends_changed(client: DRClient, username: str, known_version: int) -> bool:
    page = client.friends.list(username, known_version=known_version)
    return not page.not_modified

一页的条数可能少于 limit，甚至为 0 而仍有下一页（服务端跳过了已删除的数据），是否结束只看 next_cursor（会话消息和导出看 has_more），不要按条数判断。

自动翻页

分页的列表都有对应的遍历方法，逐项产出全部结果，只在需要下一页时才发请求：

python

from deeprespond_im import DRClient


def find_vip(client: DRClient) -> str | None:
    with client.users.iter(status="active", limit=100) as users:
        for user in users:
            if user.attributes.get("vip") == "1":
                return user.username   # 提前结束：不会再取下一页
    return None

遍历方法的参数与列表方法相同，只是没有 cursor（和 known_version）；limit 是每页的条数；
返回的迭代器也是上下文管理器，with 结束时立即释放；提前 break 不会留下进行中的请求；
某一页最终失败时抛出异常并结束遍历，已产出的项仍然有效。每一页照常自动重试；
异步客户端的遍历方法返回异步迭代器，用 async for，提前结束时用 async with 或 await it.aclose() 立即释放；
conversations.iter_messages 给出 after_seq 时向新的方向遍历，否则从最新的消息向旧的方向，整体从新到旧产出；
chatrooms.iter_members 按用户名去重：遍历期间离开又进入的用户只产出一次。

另有两个方法替你循环调用：friends.delete_all(username) 循环删除直到没有剩余的好友，返回删除的总数；moderation.process_pending(...) 逐条处理待审核的记录，见服务与方法。

批量接口的结果

批量接口（users.batch_create、friends.batch_add、messages.batch_send、messages.batch_recall 等）逐项处理，整个请求成功时返回结果对象，失败的项在结果中，不抛出异常：

python

from deeprespond_im import DRClient


def import_users(client: DRClient) -> None:
    result = client.users.batch_create([
        {"username": "u1", "nickname": "用户一"},
        {"username": "u2", "password": "S3cret-pass"},
    ])
    print(len(result.created), "个已创建")
    for item in result.failed_items:            # 失败的项：ItemError
        if item.code != "already_exists":
            print(item.key, item.code, item.message)

每个批量接口有自己的结果类型（如 BatchCreateUsersResult、BatchSendResult），字段与 REST 文档相同；
所有结果类型都有 ok（全部成功）、failed_items（失败的项，ItemError 的列表）和 raise_for_failures()（有失败的项时抛出 BatchError，否则什么也不做）；
ItemError 有 key（这一项的标识，如用户名、消息 ID）、code、message、details 和 reason；
BatchError 不是 DRError 的子类：请求本身成功了，只是有项失败。它的 failed 是失败的项，result 是整个结果；
每一项的输入是普通的 dict，类型检查器按 deeprespond_im.types 中的 TypedDict（如 CreateUserInput）检查键名和类型。

超过上限的批量

带 _all 后缀的方法（users.batch_create_all、friends.batch_add_all、messages.batch_send_all）接受任意可迭代对象（包括生成器），按每次请求的上限拆分、依次提交、合并结果，不必先把全部数据读进内存：

python

import csv
from collections.abc import Iterator

from deeprespond_im import DRClient, DRError
from deeprespond_im.types import CreateUserInput


def import_from_csv(client: DRClient, path: str) -> None:
    with open(path, newline="", encoding="utf-8") as f:
        rows: Iterator[CreateUserInput] = ({"username": row["id"], "nickname": row["name"]} for row in csv.DictReader(f))
        try:
            result = client.users.batch_create_all(rows)
        except DRError as e:
            done = e.partial   # 已完成部分的合并结果
            print("中途失败：", e.code, "已处理", len(done.created) if done else 0, "个")
            raise
    print("创建", len(result.created), "个，失败", len(result.failed), "个")

某一批整个请求失败时停止，抛出那一批原来的错误（如 RateLimitError），它的 partial 属性是已完成部分的合并结果；
应用设置了更小的每批上限、服务端以 too_many_items 拒绝时，SDK 按服务端给出的上限重新拆分；
messages.batch_send_all 的各批使用同一个 client_msg_id，整体重试也不会重复发送。

错误

调用失败时抛出 DRError 或它的子类。服务端返回的错误（见错误码）原样转为 DRError：

属性	说明
`code`	错误码，如 `not_found`、`rate_limited`，或 SDK 自己的错误码（见下文）。按它判断，不要解析 `message`
`reason`	`details.reason`，没有时为 `None`
`details`	错误响应中的 `details`，原样
`message`	说明文字，只用于日志
`status_code`	HTTP 状态码；SDK 自己的错误为 `None`
`request_id`	请求 ID，联系技术支持时提供
`retry_after`	`Retry-After` 响应头的秒数
`op`	接口名，如 `users.create`（见服务与方法的“接口名”一列）
`attempts`	这次调用一共尝试了几次
`retried`	这个错误来自自动重试后的尝试，之前的尝试可能已经执行，见重试与幂等
`partial`	只在 `_all` 方法中途失败时有：已完成部分的结果
`is_temporary`	网络错误、超时、`internal`、`5xx` 和 `rate_limited`：稍后可以再试
`is_service_unavailable`	租户或应用暂停服务、只读（`tenant_unavailable`、`app_unavailable`，以及冷却期中由它们引起的 `client_unavailable`）
`is_permanently_unavailable`	应用已删除或租户已注销

错误按 code 分为以下子类，可以按类型捕获。同一个错误码永远属于同一个子类，不认识的错误码（服务端以后新增的）是 DRError 本身：

子类	错误码
`AuthenticationError`	`unauthenticated`、`invalid_client`、`ip_not_allowed`
`PermissionDeniedError`	`permission_denied`、`user_disabled`、`not_group_member`、`group_disabled`、`chatroom_disabled`、`file_blocked`
`NotFoundError`	`not_found`、`file_expired`
`ConflictError`	`already_exists`、`version_conflict`
`LimitExceededError`	`limit_exceeded`
`RateLimitError`	`rate_limited`
`ServiceUnavailableError`	`tenant_unavailable`、`app_unavailable`
`InvalidRequestError`	`invalid_argument`、`payload_too_large`、`method_not_allowed`，以及 SDK 的 `invalid_params`
`ContentRejectedError`	`message_rejected`、`content_rejected`
`ServerError`	`internal`
`NetworkError`	SDK 的 `network_error`，同时是内置的 `ConnectionError`
`DRTimeoutError`	SDK 的 `timeout`，同时是内置的 `TimeoutError`
`InvalidResponseError`	SDK 的 `invalid_response`
`ClientUnavailableError`	SDK 的 `client_unavailable`，另有 `cause_code`

SDK 自己的错误码：

错误码	说明
`invalid_params`	参数在本地检查不通过，没有发出请求，`details.field` 指出参数
`network_error`	连接失败、连接中断、TLS 错误；底层的 `httpx` 异常在 `__cause__` 中
`timeout`	单次尝试超时，或 `deadline` 已到
`invalid_response`	响应不是预期的 JSON，如代理返回的 HTML 错误页；`details.body_prefix` 是响应的开头，`5xx` 时 `is_temporary` 为真
`client_unavailable`	客户端已关闭（`cause_code` 为 `closed`）、已永久失效，或处于换取 App Token 失败后的冷却期（`cause_code` 为最初的错误码）

python

from deeprespond_im import ConflictError, DRClient, DRError


def ensure_user(client: DRClient, username: str) -> None:
    try:
        client.users.create(username)
    except ConflictError as e:
        if e.code != "already_exists":   # 已存在的用户视为成功
            raise
    except DRError as e:
        if e.is_temporary:
            print("稍后再试", e.request_id)
        raise

异常的字符串表示含接口名、HTTP 状态、错误码、原因和 request_id，不含请求体、令牌和 Client Secret，可以直接写进日志；
所有异常都可以 pickle，Celery、multiprocessing 把它们传回调用方时字段不变；
异步任务被取消时，asyncio.CancelledError 原样向上传递，不转为 DRError。

重试与幂等

SDK 只自动重试可以安全重试的请求。每个方法属于以下一类，服务与方法的“自动重试”一列给出了每个方法的类别：

类别	说明	网络错误、超时、`5xx` 时
查询	只读的接口	自动重试
幂等	重复执行结果相同，如设置、删除、加入黑名单	自动重试
幂等键	发送消息：用 `client_msg_id` 去重，不传时 SDK 生成	用同一个 `client_msg_id` 自动重试
唯一创建	以唯一标识创建，如创建用户	自动重试；重试后得到 `already_exists` 时 `e.retried` 为真，说明可能是前一次尝试已经创建
不重试	每次执行都产生新的结果：`groups.create`、`chatrooms.create`、`media.create_upload`、`moderation.create_word_list`、`moderation.escalate`、`users.set_password`、`users.kick_all`，以及只推在线的发送（`online_only=True`）、低优先级的聊天室消息、带 `actions` 而不带 `version` 的 `moderation.decide`	只在请求确定没有发出（如连接失败）时重试

重试最多 max_retries 次（默认 3 次），间隔约 0.5、1、2 秒（各乘以 0.5 到 1.5 的随机数）；5xx 带 Retry-After 时至少等待这么久；
超过 deadline 的不再重试；
retried 的含义：自动重试后得到 already_exists、not_found、version_conflict 时，e.retried 为真，前一次尝试可能已经执行（如创建成功后响应丢失）。这时请查询确认，不要直接当作失败；
不重试的方法超时后，请先查询结果再决定是否重试，例如建群超时后按群主调用 groups.list 确认（把业务 ID 写进 attributes 便于识别）；
400、403、404、409 等错误不重试，直接抛出；
令牌失效（不带原因的 401 unauthenticated）时，SDK 换取新令牌后重发一次，不计入重试次数。

限流

收到 429 rate_limited 时（见限流），SDK 按 Retry-After 等待（加 0 到 20% 的随机量，把多个请求错开）后重发，最多 max_rate_limited_retries 次（默认 3 次），不计入 max_retries。等待的范围按 details.reason 决定：

被限流的原因	谁等待
应用的 OpenAPI 调用额度用完（没有 `details.reason`）	整个客户端：所有线程或协程的请求一起暂停，直到等待结束，不会各自重试把额度再次用完
应用每分钟的消息数（`app_message_rate`）	这个客户端的全部发送消息请求（`messages.send`、`messages.batch_send`），等待时间在 `Retry-After` 到它的 2 倍之间
举报的频率（`request_rate`）	这个客户端的全部 `moderation.report` 请求
其他原因，如某个群、某个聊天室的发送频率（`group_send_rate`、`room_send_rate`）	只有这个请求

以下情况不再等待，直接抛出 RateLimitError（e.retry_after 为服务端要求等待的秒数）：服务端要求等待的时间超过 max_retry_after（默认 60 秒）；已经重发了 max_rate_limited_retries 次；等待会超过这次调用的 deadline。客户端处于整体暂停中、而等待会超过 deadline 的调用，不发请求，直接抛出 RateLimitError。

python

import time

from deeprespond_im import DRClient, RateLimitError


def send_with_backoff(client: DRClient, username: str, text: str) -> None:
    try:
        client.messages.send(from_="notice", to_user=username, type="text", body={"text": text})
    except RateLimitError as e:
        # SDK 已经等待并重发过；仍被限流时由业务决定稍后再发
        time.sleep(e.retry_after or 1)
        raise

“整个客户端一起等待”只在一个进程内有效：多个进程同时被限流时各自等待，等待时间中的随机量会把它们错开。

本地限速

批量导入、群发通知这类后台任务，可以用本地限速控制自己的调用速率，把 OpenAPI 额度留给线上请求：

python

import os

from deeprespond_im import DRClient, RateLimit

batch_client = DRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
    rate_limit=RateLimit(per_second=300),   # 每秒最多 300 次，burst 默认等于 per_second（向上取整）
)

本地限速是令牌桶：per_second 是每秒的次数，burst 是允许的突发次数，默认 max(1, ceil(per_second))；
按项数计入服务端额度的批量接口（如批量创建用户、批量发送），本地限速也按项数扣减（每次最多扣到 burst）；
等待会超过 deadline 的，抛出 DRTimeoutError；
本地限速在一个客户端内有效，多个进程按进程数分摊额度。

上传文件

media.upload() 完成创建上传、上传内容、完成上传三步（见文件上传），返回文件信息：

python

from deeprespond_im import DRClient


def send_image(client: DRClient, path: str) -> None:
    file = client.media.upload(path, purpose="attachment", kind="image")
    client.messages.send(
        from_="notice",
        to_user="zhangsan",
        type="image",
        body={"url": file.url, "width": file.width, "height": file.height, "size": file.size},
    )

参数	说明
`source`	位置参数：文件路径（`str` 或 `pathlib.Path`）、`bytes`，或可定位的二进制文件对象（以 `rb` 打开，从当前位置读到末尾）。不能定位的流（如网络流、标准输入）不能分片重传，以 `invalid_params` 拒绝，请先写入临时文件
`purpose`、`kind`	用途和类型，如 `attachment` 和 `image`。用途与类型的组合在本地先检查
`name`	文件名，省略时取路径或文件对象的文件名。类型为 `file` 和群文件必须有文件名
`content_type`、`group_id`、`owner`	与创建上传的字段相同
`concurrency`	分片并发数，默认 4
`on_progress`	进度回调 `on_progress(uploaded, total)`（字节数）。同步客户端中可能从其他线程调用
`request_id`	追踪 ID

不超过 32 MB 的一次上传，更大的分片（每片 16 MB）并发上传。上传内容的请求直接发到对象存储，不带 App Token；
file.url 是文件地址，写进消息的 body；不要保存下载地址，它会过期；
头像（user_avatar 等）不能超过 5 MB、不能上传空文件，这些在本地先检查；
分片的上传地址过期时自动重新索取；完成上传时服务端还在合并分片的，SDK 等待后重试；分片不全的，补传后再完成；
同步客户端中按 Ctrl+C（KeyboardInterrupt）中断、异步客户端中任务被取消时，SDK 尽力取消这次上传，再把异常原样抛出。

续传大文件

分片上传（超过 32 MB）中断后，24 小时内可以用 media.resume_upload(file_id, source) 和同一份内容继续：SDK 查询已上传的分片，只补传缺少的，再完成上传。upload() 抛出的异常中不带文件 ID，需要续传的大文件请自己创建上传、保存 file_id，再用 resume_upload() 上传内容：

python

import os

from deeprespond_im import DRClient
from deeprespond_im.types import File


def upload_big_file(client: DRClient, path: str, saved_file_id: str | None) -> File:
    if saved_file_id is None:
        session = client.media.create_upload(
            purpose="attachment", kind="file", size=os.path.getsize(path), name=os.path.basename(path)
        )
        saved_file_id = session.file_id   # 保存下来，中断后用它续传
    return client.media.resume_upload(saved_file_id, path)

下载文件

media.download(url, dest) 用文件地址换取下载地址，流式写入路径或可写的二进制文件对象，不把整个文件读进内存，返回写入的字节数。下载地址过期（对象存储返回 403）时重新换取一次：

python

from deeprespond_im import DRClient


def save_attachment(client: DRClient, file_url: str) -> int:
    return client.media.download(file_url, "/tmp/attachment.bin")

只需要下载地址（如交给浏览器下载）时，用 media.get_download_urls([url, ...])。

应用和租户的状态

应用被停用、暂停，租户被暂停或欠费时（见应用和租户的状态），调用的结果如下：

状态	SDK 的表现	建议
只读	查询和处置类操作正常；新增数据、修改资料的写操作抛出 `ServiceUnavailableError`（`app_unavailable`，欠费时 `reason` 为 `arrears`）	告警；暂停写入类的后台任务
应用停用或暂停、租户暂停	已签发的令牌失效，SDK 重新换取时得到 `app_unavailable` 或 `tenant_unavailable`，抛出 `ServiceUnavailableError`；之后 2 到 5 分钟内不再换取，调用直接抛出 `ClientUnavailableError`（`cause_code` 为最初的错误码）	告警，在控制台查看状态；冷却期结束后 SDK 自动再试，恢复后不必重启
Client Secret 错误或已吊销、IP 不在白名单	抛出 `AuthenticationError`（`invalid_client`、`ip_not_allowed`）；之后 1 分钟内调用直接抛出 `ClientUnavailableError`	更新凭据后调用 `client.update_credentials()`，立即结束冷却期
应用已删除、租户已注销	抛出 `AuthenticationError`（`reason` 为 `app_deleted`、`tenant_closed`），客户端永久失效，之后都抛出 `ClientUnavailableError`	清除保存的凭据，停止调用

按“服务暂停”处理的代码，用 e.is_service_unavailable 判断，它同时涵盖 ServiceUnavailableError 和冷却期中的 ClientUnavailableError：

python

from deeprespond_im import DRClient, DRError


def sync_profile(client: DRClient, username: str, nickname: str) -> bool:
    try:
        client.users.update(username, nickname=nickname)
    except DRError as e:
        if e.is_service_unavailable or e.is_permanently_unavailable:
            print("IM 服务不可用，稍后同步：", e)
            return False
        raise
    return True

日志与排查

SDK 用标准库 logging 记录日志，日志器为 deeprespond_im，下分 deeprespond_im.http（请求、重试、上传）、deeprespond_im.token（App Token 的换取）和 deeprespond_im.callback（接收回调）。SDK 不添加处理器，是否输出由你的日志配置决定：

重试、限流等待记 WARNING 或 DEBUG；最终失败的网络错误、超时、5xx、鉴权失败记 ERROR；其他 4xx 是业务结果，记 DEBUG；
日志不含请求体、App Token 和 Client Secret；
httpx 自己在 INFO 级别为每个请求记一行完整的地址，对象存储的上传地址、下载地址中的签名都在查询参数里。SDK 默认给 httpx 日志器加了过滤器，把地址的查询参数替换为 ?…（redact_httpx_logs=False 关闭）。httpcore 日志器在 DEBUG 级别记录响应头，生产环境请把它设为 WARNING 以上。

钩子

hooks 选项在每次尝试前后调用你的函数，可以注入追踪头、记录指标。after_response 收到 RequestInfo：接口名 op（换取令牌为 app.token）、method、url（不含查询参数）、第几次尝试 attempt、status_code、错误码 code 和 reason、request_id、耗时 elapsed 和这次尝试前的限流等待 rate_limit_wait：

python

import logging
import os

import httpx

from deeprespond_im import DRClient, Hooks, RequestInfo

log = logging.getLogger("im")


def add_trace(request: httpx.Request) -> None:
    request.headers["traceparent"] = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"


def record(info: RequestInfo) -> None:
    log.info("%s attempt=%d status=%s code=%s %.0fms", info.op, info.attempt, info.status_code, info.code, info.elapsed * 1000)


client = DRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
    hooks=Hooks(before_request=add_trace, after_response=record),
)

钩子在发出请求的线程中同步调用，不要做耗时的操作；钩子抛出的异常原样向上传递；
不要在钩子中记录 Authorization 请求头和请求体；
异步客户端用 AsyncHooks，两个钩子可以是普通函数，也可以是协程函数。

统计

client.stats() 返回按接口名统计的 OpStats：尝试次数 requests、最终失败的调用 failures、重试和限流后重发的次数 retries、限流等待的总时长 rate_limit_wait（秒）。可以定期采集到监控系统：

python

from deeprespond_im import DRClient


def report_stats(client: DRClient) -> None:
    for op, s in client.stats().items():
        print(op, s.requests, s.failures, s.retries, round(s.rate_limit_wait, 2))

联系技术支持时，请提供错误的 request_id、op、code 和发生的时间。

在 asyncio 中使用

FastAPI、Starlette、aiohttp 等异步程序使用 AsyncDRClient，不要在异步代码中调用同步客户端（它会阻塞事件循环）。在 FastAPI 中，可以在应用的生命周期中创建和关闭客户端：

python

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from deeprespond_im import AsyncDRClient

im = AsyncDRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    yield
    await im.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/im/ticket")
async def im_ticket(user_id: str) -> dict[str, str]:
    ticket = await im.users.issue_login_ticket(user_id, auto_create=True)
    return {"ticket": ticket.ticket}

异步客户端可以在模块级创建，但只能在第一次发出请求时所在的事件循环中使用。在其他事件循环中使用（如第二次 asyncio.run()、pytest-asyncio 为每个用例新建的事件循环）时抛出 RuntimeError，请每个事件循环各建一个；
遍历用 async for：async for user in im.users.iter(status="active"): ...；
可以用 asyncio.timeout() 限制整个调用，取消会中断正在进行的请求和等待；也可以用方法的 deadline 参数。被取消或超时的写操作可能已经执行，按重试与幂等中的类别处理；
异步客户端的 App Token 存储用 AsyncTokenStore（如 deeprespond_im.redis.AsyncRedisTokenStore 和 redis.asyncio.Redis），钩子用 AsyncHooks；
同步客户端中没有取消，只能用 timeout、deadline 和 with_options() 限制时长。

测试自己的代码

deeprespond_im.testing.FakeOpenAPI 是一个假的 OpenAPI，不需要网络：按接口名设置返回值或错误，检查你的代码发出了什么请求。

python

from deeprespond_im import RateLimitError
from deeprespond_im.testing import FakeOpenAPI


def test_send_notice() -> None:
    fake = FakeOpenAPI()
    fake.respond("messages.send", status=201, json={"message_id": "m1", "client_msg_id": "order-1", "created_at": "2026-10-06T08:00:00Z"})
    client = fake.client()          # 连到假的 OpenAPI 的 DRClient；fake.async_client() 返回 AsyncDRClient

    result = client.messages.send(from_="notice", to_user="zhangsan", client_msg_id="order-1", type="text", body={"text": "hi"})

    assert result.message_id == "m1" and not result.duplicate
    assert fake.calls("messages.send")[0].json["to_user"] == "zhangsan"
    client.close()


def test_rate_limited() -> None:
    fake = FakeOpenAPI()
    fake.fail("users.get", "rate_limited", retry_after=120, times=None)   # 一直返回 429
    client = fake.client()
    try:
        client.users.get("zhangsan")
    except RateLimitError as e:
        assert e.retry_after == 120          # 超过 max_retry_after，不等待
    else:
        raise AssertionError("应当抛出 RateLimitError")
    client.close()

方法	说明
`fake.respond(op, json=..., status=200, headers=..., times=None)`	设置接口的返回值；`times` 为生效的次数，默认一直有效；后设置的先用
`fake.fail(op, code, status=..., details=..., retry_after=..., times=1)`	让接口返回错误，HTTP 状态码按错误码自动选择，默认一次
`fake.handle(op, responder, times=None)`	用自己的函数处理请求，返回 `httpx.Response`；也可以抛出 `httpx` 的异常模拟网络错误
`fake.revoke_tokens()`	让已换取的 App Token 失效，测试令牌失效后的重新换取
`fake.requests`、`fake.calls(op)`	收到的请求：接口名、方法、路径、查询参数、请求头、请求体，`.json` 是解析后的请求体
`fake.client(kwargs)`、`fake.async_client(kwargs)`	连到它的客户端，其他参数原样传给客户端（如 `max_retries=0`）

op 是接口名，见服务与方法的“接口名”一列，如 users.list_mutes 的接口名是 users.mutes；
没有设置的接口返回 200 和 {}（没有结果的返回 204）。响应中缺少的字段解析为零值（空字符串、0、False、None），请在 json 中给出测试要检查的字段；
状态码有含义的接口要按 REST 文档设置 status，如发送消息新写入时为 201，返回 200 表示重复提交（duplicate 为真）；
使用 pytest 时，可以在 conftest.py 中启用夹具 pytest_plugins = ["deeprespond_im.testing.pytest"]，得到 im_fake、im_client、im_async_client（后者需要 pytest-asyncio）。

测试接收回调的代码见本地测试。

使用独立频道

python

created = client.channels.create("meeting-demo", name="项目会议", media="video", access_mode="ticket")
issued = client.channels.issue_tickets(created.channel.channel_id, users=[{"username": "alice", "role": "publisher"}])
# 只将 issued.tickets 中目标用户的票据交付该用户，勿记录原文。
page = client.channels.list(status="open", limit=20)
corrections = client.channels.usage_adjustments(limit=20)

AsyncDRClient 使用相同服务和方法，调用加 await。创建为 ChannelsCreateResult，查询为 RTCChannelResult，票据为具体 tickets 包装；字段保留类型和原始数据。expires_at 用 UNSET / None / 值表示不传 / 清除 / 设置；stats 使用关键字 from_、to。分页按 next_cursor 手动继续，没有频道 iter 方法。签票据不自动重试，移出和撤票固定 operation_id。见频道方法，本轮新制品尚未发布。

调用接口 ​

命名规则 ​

不传、传 null 与传值 ​

原样发送 JSON ​

每个方法的通用参数 ​

返回值 ​

分页 ​

自动翻页 ​

批量接口的结果 ​

超过上限的批量 ​

错误 ​

重试与幂等 ​

限流 ​

本地限速 ​

上传文件 ​

续传大文件 ​

下载文件 ​

应用和租户的状态 ​

日志与排查 ​

钩子 ​

统计 ​

在 asyncio 中使用 ​

测试自己的代码 ​

使用独立频道 ​

调用接口

命名规则

不传、传 null 与传值

原样发送 JSON

每个方法的通用参数

返回值

分页

自动翻页

批量接口的结果

超过上限的批量

错误

重试与幂等

限流

本地限速

上传文件

续传大文件

下载文件

应用和租户的状态

日志与排查

钩子

统计

在 asyncio 中使用

测试自己的代码

使用独立频道