跳转至

调用接口

调用路径是 feishu.rest.<业务域>.<版本>.<资源>.<方法>,每个方法都有同步和 async_ 开头的异步两种形式。

传参

路径参数按位置传,查询参数和请求体字段按关键字传。请求体也可以整个通过 data 传字典或模型:

from larkify import AppAuthStrategy, Feishu
from larkify.models.im.v1.message import CreateMessageBody

feishu = Feishu(AppAuthStrategy("cli_xxx", "app_secret"))

feishu.rest.im.v1.message.get("om_xxx", user_id_type="open_id")

feishu.rest.im.v1.message.create(
    receive_id_type="open_id",
    data={"receive_id": "ou_xxx", "msg_type": "text", "content": "{}"},
)
feishu.rest.im.v1.message.create(
    receive_id_type="open_id",
    data=CreateMessageBody(receive_id="ou_xxx", msg_type="text", content="{}"),
)

请求体在发送前由 pydantic 校验,缺少必填字段会在本地报错。Feishu(..., rest_api_validate_body=False) 可以关闭校验。文档里没有的字段不会被拦截,会原样发送。

字段名和 Python 关键字或 pydantic 保留名冲突时加下划线,例如 from_;传字典时使用接口原本的字段名。极少数接口的请求体字段和查询参数同名(如 drive.v1.permission_member 的 type),这时请求体只能走 data。

响应

Response 包装了 httpx2.Response:

属性 说明
parsed_data 校验后的 data,下载接口为 bytes
code / msg 业务状态码和信息
log_id 向飞书支持反馈问题时使用的日志 ID
json() 原始 JSON,parsed_data 校验不过时的兜底
file_name 下载文件名
raw_request / raw_response 底层 httpx2 对象

响应模型的字段都是 T | None,文档没写的字段保留在 model_extra 里。

分页

for chat in feishu.rest.paginate(feishu.rest.im.v1.chat.list, page_size=50):
    print(chat.name)
async for chat in feishu.rest.paginate(feishu.rest.im.v1.chat.async_list):
    print(chat.name)

元素类型会被推导出来。响应里有多个列表、无法确定取哪一个时,用 paginate_with 自行挑选,第二个参数之后是传给接口的参数:

users = feishu.rest.paginate_with(
    feishu.rest.contact.v3.group_member.simplelist,
    lambda response: response.parsed_data.memberlist or [],
    "group_id",
)

上传与下载

resp = feishu.rest.im.v1.image.create(
    image_type="message",
    image=("cat.png", open("cat.png", "rb"), "image/png"),
)
image_key = resp.parsed_data.image_key

resp = feishu.rest.im.v1.image.get(image_key)
content = resp.parsed_data

# 大文件用流式,不会读入内存
resp = feishu.rest.drive.v1.file.download("file_token", stream=True)
with open(resp.file_name or "download", "wb") as f:
    for chunk in resp.iter_bytes():
        f.write(chunk)

未收录的接口

resp = feishu.request(
    "GET",
    "/open-apis/unified_kms/v1/autonomous_keys",
    token_types=("tenant",),
)
print(resp.json())
resp = await feishu.arequest(
    "GET",
    "/open-apis/unified_kms/v1/autonomous_keys",
    token_types=("tenant",),
)
print(resp.json())

feishu.request 同样会校验响应、处理 token 和重试。个别接口的字段不在 data 里,比如获取机器人信息,用 data_key=None 取整个响应:

resp = await feishu.arequest("GET", "/open-apis/bot/v3/info", data_key=None)
print(resp.parsed_data["bot"]["open_id"])

timeout 可以覆盖客户端的超时配置,None 表示不限时。它对每次重试分别生效,也用于获取这次调用需要的 token:

resp = feishu.request("GET", "/open-apis/unified_kms/v1/autonomous_keys", timeout=120)