seatable-api-async 0.3.0

异步版本的 SeaTable API Python 客户端，基于 aiohttp 实现

SeaTable API Async

异步版本的 SeaTable API Python 客户端库，基于 seatable-api-python 改造，使用 aiohttp 和 async/await 实现完全异步操作。

特性

• 完全异步 - 基于 aiohttp 和 async/await，支持高并发操作
• WebSocket 支持 - 内置 SocketIO 客户端，支持实时事件推送
• 自动 Token 管理 - JWT token 自动刷新，无需手动干预
• 批量操作 - 支持批量增删改，提高数据处理效率
• 分块操作 - 自动处理大数据集，突破 SeaTable API 单次请求限制（新增 v0.3.0）
• 完整功能 - 覆盖表、行、列、视图、链接、评论、文件上传等所有核心 API

安装

# 使用 pip
pip install seatable-api-async

# 使用 uv
uv add seatable-api-async

快速开始

AccountApi 操作

import asyncio
from seatable_api_async import AccountApiAsync

async def main():
    async with AccountApiAsync(
        login_name="[e-mail]",
        password="your_password",
        server_url="https://cloud.seatable.io"
    ) as account:
        # 列出工作区
        workspaces = await account.list_workspaces()

        # 获取 Base API Token
        token = await account.get_temp_api_token(
            workspace_id=1,
            dtable_name="My Base"
        )

if __name__ == "__main__":
    asyncio.run(main())

SeatableApi 操作

from seatable_api_async import SeaTableApiAsync

async def main():
    async with SeaTableApiAsync(
        token="your_api_token",
        server_url="https://cloud.seatable.io"
    ) as base:
        # 列出行
        rows = await base.list_rows("Table1")

        # 添加行
        await base.append_row("Table1", {"Name": "Alice", "Age": 30})

        # 批量添加
        await base.batch_append_rows("Table1", [
            {"Name": "Bob", "Age": 25},
            {"Name": "Charlie", "Age": 35}
        ])

        # SQL 查询
        results = await base.query("SELECT * FROM Table1 WHERE Age > 25")

if __name__ == "__main__":
    asyncio.run(main())

WebSocket 实时通讯

from seatable_api_async import SeaTableApiAsync, SocketIOAsync

async def main():
    async with SeaTableApiAsync(token, server_url) as base:
        # 使用 async with 自动管理连接
        async with SocketIOAsync(base) as socket:
            @socket.on("custom_event")
            async def handle_event(data):
                print(f"Received: {data}")

            await socket.emit("my_event", {"data": "hello"})
            # 退出时自动断开连接

if __name__ == "__main__":
    asyncio.run(main())

API 文档

官方文档

• SeaTable API 英文文档: https://api.seatable.com/
• SeaTable API 中文文档: https://docs.seatable.cn/published/seatable-api/home.md

使用说明

详细的 API 方法请参考同步版本 seatable-api 文档，所有方法名称保持一致，只需添加 await 关键字。

主要模块：

• AccountApiAsync - 账户管理、工作区操作
• SeaTableApiAsync - Base 操作、表/行/列/视图管理、SQL 查询
• SocketIOAsync - WebSocket 实时通讯

API 限制

• 批量操作（batch_append/update_rows）：最多 1,000 行/请求
• 批量删除（batch_delete_rows）：最多 10,000 行/请求
• 列出行（list_rows）：最多 1,000 行/请求
• SQL 查询返回：最多 10,000 行
• 速率限制：SeaTable Cloud 200 请求/分钟（Base 操作）

开发建议

使用异步上下文管理器

async with SeaTableApiAsync(token, server_url) as api:
    rows = await api.list_rows("Table1")
# 会话自动关闭

批量操作优先

# 推荐 - 使用批量操作
await api.batch_append_rows("Table1", rows)

# 避免 - 循环单个操作
for row in rows:
    await api.append_row("Table1", row)  # 性能较差

处理大数据集（分块操作）

当数据量超过 SeaTable API 限制时，使用 _chunked 方法自动分块处理：

# 添加超过 1000 行的数据
rows = [{"Name": f"User {i}", "Age": 20} for i in range(5000)]
await api.batch_append_rows_chunked("Table1", rows, chunk_size=1000)

# 获取所有行（自动分页）
all_rows = await api.list_rows_chunked("Table1", max_rows=10000)

# 更新大量数据
updates = [{"row_id": row["_id"], "row": {"Status": "Processed"}} for row in all_rows]
await api.batch_update_rows_chunked("Table1", updates)

# 删除大量行
row_ids = [row["_id"] for row in all_rows]
await api.batch_delete_rows_chunked("Table1", row_ids)

# 使用进度回调监控进度
def progress_callback(processed, total):
    print(f"进度: {processed}/{total} ({processed*100//total}%)")

await api.batch_append_rows_chunked(
    "Table1",
    large_dataset,
    progress_callback=progress_callback
)

可用的分块方法：

• batch_append_rows_chunked() - 添加大量行（>1000）
• batch_update_rows_chunked() - 更新大量行（>1000）
• batch_delete_rows_chunked() - 删除大量行（>10000）
• list_rows_chunked() - 获取所有行（>1000）
• batch_add_links_chunked() - 添加大量链接（>1000）
• batch_remove_links_chunked() - 删除大量链接（>1000）
• batch_update_links_chunked() - 更新大量链接（>1000）

测试

# 配置环境变量
cp .env_template .env

# 运行测试
pytest tests/

贡献

欢迎贡献代码！请 Fork 本仓库，创建特性分支，提交 Pull Request。

许可证

本项目基于 Apache License 2.0 开源协议。

链接

• PyPI
• GitHub
• SeaTable API 官方文档
• SeaTable API 官方介绍
• SeaTable API Python (同步版本)
• SeaTable 编程手册(同步版本)

致谢

本项目基于 seatable-api，感谢 SeaTable 团队的优秀工作。