Skip to content

bo-john/seatable-api-async

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

11 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

SeaTable API Async

PyPI version Python License

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

特性

  • 完全异步 - 基于 aiohttpasync/await,支持高并发操作
  • WebSocket 支持 - 内置 SocketIO 客户端,支持实时事件推送
  • 自动 Token 管理 - JWT token 自动刷新,无需手动干预
  • 批量操作 - 支持批量增删改,提高数据处理效率
  • 完整功能 - 覆盖表、行、列、视图、链接、评论、文件上传等所有核心 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="your_email@example.com",
        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 文档

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

主要模块:

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

开发建议

使用异步上下文管理器

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)  # 性能较差

测试

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

# 运行测试
pytest tests/

贡献

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

许可证

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

链接

致谢

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

About

将 seatable-api 核心功能由同步改成异步

Resources

License

Stars

Watchers

Forks

Packages

 
 
 

Contributors

Languages