Baserow WebSocket API 实践指南：实时协作协议、页面订阅与自托管接入

Baserow WebSocket API 实践指南实时协作协议、页面订阅与自托管接入【免费下载链接】baserowBuild databases, automations, apps agents with AI — no code. Open source platform available on cloud and self-hosted. GDPR, HIPAA, SOC 2 compliant. Best Airtable alternative.项目地址: https://gitcode.com/GitHub_Trending/ba/baserowBaserow 的 WebSocket API 是其实时协作的通信基础当工作区内发生数据变化时后端通过 WebSocket 把变更广播给所有在线协作者前端据此免刷新地更新已加载的数据。本文基于官方文档 docs/apis/web-socket-api.md 完整讲解连接认证、消息格式、Web Socket ID 去回显机制、页面级订阅与全部消息类型清单并结合 CoreConsumer 等后端源码揭示协议背后的实现链路帮助你为自己的 Baserow 集成自建应用、第三方同步工具、监控面板等构建一个行为正确、低冗余的 WebSocket 客户端。连接与认证JWT 鉴权的 WebSocket 握手连接 WebSocket 前必须先通过 REST API 完成认证并获取一个 JSON Web TokenJWT然后将其作为查询参数传入 WebSocket 地址wss://api.baserow.io/ws/core/?jwt_tokenYOUR_JWT_TOKEN如果你部署的是自托管实例把api.baserow.io替换为自己的后端域名即可。连接建立后该连接只会收到发给「当前认证用户所属于的工作区」的消息即消息投递在用户维度天然做了隔离。JavaScript 下最简的连接方式如下const socket new WebSocket(wss://api.baserow.io/ws/core/?jwt_tokenYOUR_JWT_TOKEN) socket.onopen () { console.log(The connection is made) } socket.onmessage (message) { console.log(Received, message) }源码视角路由与认证中间件从源码结构看ws/core/路径被直接绑定到核心的CoreConsumer见 backend/src/baserow/ws/routing.pywebsocket_urlpatterns [re_path(r^ws/core/, CoreConsumer.as_asgi())]真正解析jwt_token查询参数的是 JWTTokenAuthMiddleware。它在 ASGI scope 上注入两个关键属性scope[user]调用get_user(token)解析 JWT。缓存命中的已认证用户可走快速路径免数据库查询匿名令牌anonymous在未被DISABLE_ANONYMOUS_PUBLIC_VIEW_WS_CONNECTIONS禁用时会返回AnonymousUser用于公开分享视图的匿名实时连接。scope[web_socket_id]若客户端在 URL 上显式带了web_socket_id查询参数则复用之否则自动生成一个随机 UUID——这正是后文「Web Socket ID」机制的来源。也就是说官方文档中的「用 JWT 换连接」在实现上是认证失败scope[user] is None的连接会在CoreConsumer.connect中先收到一条success: false的authentication消息随后被服务端close()。认证握手消息与 Web Socket ID连接建立后客户端会先收到一条authentication消息指示 JWT 认证是否成功。若成功消息中会携带一个web_socket_id{ type: authentication, success: true, web_socket_id: 934254ab-0c87-4dbc-9d71-7eeab029296c }对照 CoreConsumer.connect 的实现当前版本实际发出的认证消息还包含一个replay_enabled字段取决于用户是否已认证以及服务端是否开启了实时事件记录用于告知客户端重连时能否走「事件重放」而非强制刷新——这是文档未展开但对自建客户端有用的握手信息。用 WebSocketId 头排除自己的回显消息这个 ID 的核心用途是去回显当你通过 REST API 发起变更例如更新应用名称时可以在 HTTP 请求中带上WebSocketId头服务端就会在广播该变更时跳过你这个连接因为这条消息反映的本来就是你自己刚执行的变更PATCH /api/applications/1/ Host: api.baserow.io Content-Type: application/json WebSocketId: 934254ab-0c87-4dbc-9d71-7eeab029296c { name: Test }在源码中该头最终转化为所有广播任务的ignore_web_socket_id参数。以 Celery 任务 broadcast_to_group 为例它先查出工作区内的全部用户 ID再交给broadcast_to_users统一投递而消费端 broadcast_to_users / broadcast_to_group 每个处理器都会先比较ignore_web_socket_id ! web_socket_id不匹配才send_json。测试用例 backend/tests/baserow/ws/test_ws_consumers.py 中专门验证了「被忽略的 WebSocket 收不到自己的广播」这一行为可作为协议行为的回归依据。实时更新消息统一的 JSON 信封广播的实时消息永远是 JSON且总是包含一个type键标明发生了什么变化。例如type为application_created时会附带一个application键其中是新建应用的序列化结果。官方文档给出的示例消息如下另一位用户在你也所属的工作区中创建了数据库应用{ type: application_created, application: { id: 123, name: Test, order: 8, type: database, workspace: { id: 1, name: Brams workspace }, tables: [] } }值得一提的是这条消息的生成方式Celery 任务 broadcast_application_created 会先对「工作区内所有满足读权限的用户」逐一做权限检查然后按用户上下文分别序列化应用因为不同用户可见的内容可能不同最后通过broadcast_to_users_individual_payloads用一条 channel-layer 消息把payload_mapuser_id → payload批量发出去。消费端的 broadcast_to_users_individual_payloads 只把payload_map中属于当前连接用户的 payload 发回给该连接——即文档中「只收到自己所属工作区的消息」在实现上是通过用户 ID 过滤保证的。页面级订阅Table 页与 Row 页默认情况下用户会收到所有与工作区和应用相关的核心消息但针对具体「页面」如某个表格页的高频消息需要显式订阅以避免消息过载。一条连接可以同时订阅多个页面且必须手动退订才会停止接收。Table 页订阅订阅 table 页后你将收到与某张 Baserow 表格相关的更新新行创建、行更新等。table 页要求table_id参数{ page: table, table_id: 1 }订阅成功后收到确认消息{ type: page_add, page: table, parameters: { table_id: 1 } }源码中对应 TablePageTypetype tableparameters [table_id]。有两点值得注意权限校验CoreConsumer._add_page_scope会先通过run_database_sync调用page_type.can_add(...)。对 table 页can_add要求表格存在、且用户对该表格拥有listen_to_all_database_table_events操作权限校验不过则静默拒绝不会加入任何组也就收不到page_add确认。底层组名get_group_name返回table-{table_id}同时该页还会加入权限组permissions-table-{table_id}——当你的表权限被撤销时服务端可通过权限组主动把你踢出对应页面组消费端由 users_removed_from_permission_group 处理器实现。Row 页订阅订阅 row 页会收到与某一行相关的额外更新典型场景是行历史row history的变更。注意行删除这类变更应通过上面的 table 页订阅获取row 页只覆盖行粒度的增量信息。row 页要求table_id与row_id两个参数{ page: row, table_id: 1, row_id: 1 }确认消息同样以page_add形式返回{ type: page_add, page: row, parameters: { table_id: 1, row_id: 1 } }对应实现 RowPageTypecan_add除表格权限外还会调用RowHandler.get_row(user, table, row_id)确认行存在且可读其 channel 组名为table-{table_id}-row-{row_id}。行相关的广播 payload如rows_created、rows_updated在 backend/src/baserow/contrib/database/ws/rows/messages.py 中集中定义包含table_id与序列化后的行数据。退订页面停止接收某页面的更新时发送一条remove_page消息参数与订阅时一致{ remove_page: row, table_id: 1, row_id: 1 }从 _remove_page_scope 看退订流程会把连接移出页面组与若不再被其他页面需要时权限组移除SubscribedPages记录并向客户端回发确认{ type: page_discard, page: row, parameters: { table_id: 1, row_id: 1 } }完整消息类型清单官方文档按作用域列出了全部广播消息类型这里完整继承数据库与 Premium 部分以插件/付费模块形式注册但消息类型对客户端一视同仁核心消息类型authenticationpage_addpage_discardbefore_group_deleteduser_updateduser_deleteduser_restoreduser_permanently_deletedgroup_createdgroup_updatedgroup_deletedgroup_restoredgroup_user_addedgroup_user_updatedgroup_user_deletedapplication_createdapplication_updatedapplication_deletedapplications_reorderedDatabase 消息类型table_createdtable_updatedtable_deletedtables_re_orderedfield_createdfield_updatedfield_deletedfield_restoredrows_createdrows_updatedrows_deletedbefore_rows_updatebefore_rows_deleterow_history_updatedview_createdview_updatedview_deletedview_filter_createdview_filter_updatedview_filter_deletedview_filter_group_createdview_filter_group_updatedview_filter_group_deletedview_sort_createdview_sort_updatedview_sort_deletedview_decoration_createdview_decoration_updatedview_decoration_deletedview_field_options_updatedviews_reorderedPremium 消息类型row_comment_createdrow_comment_updatedrow_comment_deleted底层实现速览从 REST 变更到客户端接收如果你需要理解「一条广播是怎么送达的」仓库内部文档 docs/technical/websockets.md 与源码给出了完整链路消费端框架Baserow 基于 Django Channels 的AsyncJsonWebsocketConsumer实现 CoreConsumer负责全部前端连接、后端事件与消息收发channel layer 使用RedisChannelLayer做跨进程通信。页面注册表每种可订阅页面都是一个 PageType 实例声明type、parameters、can_add、get_group_name等统一注册到page_registry。插件可以注册新页面类型而无需改动 consumer 本身——这解释了为何 table/row 页的实现放在 contrib/database/ws/pages.py 而非核心 ws 模块。广播任务业务信号层不直接触碰 channel layer而是调用 backend/src/baserow/ws/tasks.py 中的 Celery 任务broadcast_to_users、broadcast_to_channel_group、broadcast_to_permitted_users等。后者还会在执行前做权限计算CoreHandler().check_permission_for_multiple_actors只把 payload 投递给有权限的用户。可靠性层补充知识服务端可选地把可重放事件持久化到ws_realtime_events表客户端断线重连时发送replay_events消息携带last_seen_id游标拉回漏掉的更新超过上限配置项BASEROW_REALTIME_REPLAY_MAX_EVENTS默认 200则返回force_refreshtrue提示客户端刷新。握手消息中的replay_enabled字段即为该能力的开关信号详见 docs/technical/websockets.md。测试侧可以重点关注 backend/tests/baserow/ws/ 目录test_ws_consumers.py覆盖连接握手与页面订阅test_ws_auth.py覆盖 JWT 鉴权与匿名连接test_ws_replay_isolation.py与test_ws_dispatch_isolation.py则验证重放与投递的隔离行为。自建客户端落地要点综合文档与源码自建一个符合 Baserow 协议的 WebSocket 客户端应遵循以下约定通过 REST 认证获取 JWT以jwt_token查询参数连接wss://你的后端/ws/core/收到authentication消息后确认success为true并记下web_socket_id与replay_enabled所有后续 REST 写请求带上WebSocketId头避免收到自己的回显广播进入表格/行详情等重数据页面时再发送page订阅消息离开时发送remove_page退订以type字段作为消息分发入口对照上文三类清单处理各业务更新断线重连后如replay_enabled为真可发replay_events并携带最后处理的_event_id根据返回的force_refresh决定是否整体刷新。以上行为均有对应源码与测试用例可查证CoreConsumer、JWTTokenAuthMiddleware、broadcast 任务集适合作为集成验收时的行为基线。【免费下载链接】baserowBuild databases, automations, apps agents with AI — no code. Open source platform available on cloud and self-hosted. GDPR, HIPAA, SOC 2 compliant. Best Airtable alternative.项目地址: https://gitcode.com/GitHub_Trending/ba/baserow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考