Frappe 实时服务器自定义事件处理指南为自定义应用编写 Realtime Handlers【免费下载链接】frappeLow code web framework for real world applications, in Python and Javascript项目地址: https://gitcode.com/GitHub_Trending/fr/frappe导读Frappe 框架内置了一套基于 Python Socket.IO 的实时Realtime服务器用于把 Redis 中发布的事件桥接推送到浏览器端连接。本指南面向需要在自定义应用中增加实时事件能力的开发者从 handler 的存放位置、编写语法、装饰器选项到权限检查与事件推送的完整闭环逐一给出可直接复用的代码与底层实现依据。阅读完成后你将能够为自己的应用注册实时事件处理器并在 Web 进程中向指定用户、文档、DocType 或房间推送实时消息。实时服务器的架构概览Frappe 的实时服务器是一套独立的 Python Socket.IO 服务python -m frappe.realtime.server运行在 asyncio/uvicorn 之上与 Web/gunicorn 进程完全分离。它通过订阅 Redis 的events频道接收 Web 进程发布的事件再转发给已连接的 Socket.IO 客户端见 bridge.py 中的RedisBridge实现。从源码看整个实时链路可以概括为Web 进程调用publish_realtime或其命名辅助函数把事件写入 Redis 的events频道消息结构为{event, message, room, namespace}实时服务器通过RedisBridge订阅该频道并解析消息按room在对应的/{site}命名空间下向客户端emit浏览器客户端socketio_client.js通过frappe.realtime.on(...)注册回调、通过socket.emit(...)发送事件。本指南面向的是需要在这条链路上新增自定义事件的应用开发者——你不需要修改frappe/realtime目录下的任何文件只需在自己应用的代码里注册 handler 即可。启动与嵌入方式独立运行python -m frappe.realtime.server入口见 server.py 的main()与serve()。进程内嵌入构建RealtimeServer并在自有的线程中调用run()server RealtimeServer() threading.Thread(targetserver.run, daemonTrue).start() # ... server.stop()RealtimeServer在构造时会先load_handlers再wire确保所有realtime.on注册先于事件绑定完成有对应单测验证见 test_realtime_py.py 中test_handlers_are_loaded_before_events_are_wired。服务器的端口、Redis 地址等参数全部来自common_site_config.json/site_config.json不读取环境变量见 config.py配置项默认值说明socketio_port9000实时服务器监听端口redis_queueredis://127.0.0.1:11311桥接订阅使用的 Redis 队列地址socketio_uds无指定 Unix Domain Socket 路径时优先使用 UDS 绑定socketio_worker_threads4处理阻塞型 handler 的工作线程池大小客户端较多时建议调大default_site无Host 为 localhost/127.0.0.1 时解析站点名的回退值webserver_port/webserver_host无开发模式下回叫 Web 进程时的地址覆盖你的 Handler 应该放在哪里应用的自定义 handler 统一放在your_app/your_app/realtime/handlers.py服务器在启动时会遍历站点上安装的每个应用并尝试导入app.realtime.handlers模块核心实现见 registry.py 的discover_app_handlers模块不存在应用只是没有实时 handler完全正常不会报错模块存在但导入抛出异常服务器会在启动阶段响亮地失败绝不吞掉错误——一个损坏的 handler 文件会立刻暴露出来。这个响亮失败的设计在源码中有明确注释并且在 test_realtime_py.py 的TestDispatch/ 启动顺序相关测试中得到了验证。编写一个 Handler最基础的 handler 写法如下from frappe.realtime import Socket, realtime realtime.on(project_subscribe) async def project_subscribe(socket: Socket, project: str) - None: if await socket.has_permission(Project, project): await socket.join(fproject:{project})关键约定事件名project_subscribe就是浏览器端用frappe.realtime.on(...)订阅、客户端用socket.emit(...)发送时使用的事件名第一个参数永远是类型化的Socket其余参数按位置对应客户端发送的 payload装饰器原样返回函数因此它只是一个普通函数你可以直接调用它或对它做单元测试。async 还是普通函数推荐写成async def它们在事件循环上运行用await调用 socket 方法await socket.join(...)、await socket.has_permission(...)普通def也可以工作它会在线程池的工作线程中运行并拿到一个阻塞式的SyncSocketAPI 与Socket相同但无需await。当函数体必须阻塞时例如使用了没有 async 客户端的库、或需要frappe_context使用普通函数。阻塞型 handler 由 dispatch.py 的_call通过asyncio.to_thread调度到工作线程SyncSocketsocket.py则通过run_coroutine_threadsafe把协程提交回事件循环等待完成并带有 120 秒的兜底超时LOOP_CALL_TIMEOUT。装饰器选项realtime.on( project_subscribe, frappe_contextFalse, # 打开 Frappe 上下文DB session供 handler 体使用 allow_guestFalse, # 为 False 时socket.user Guest 的事件会被丢弃 )默认值均为False。在 registry.py 的Registry.on中两个标志与事件名、函数、所属应用一起被封装进Handler数据类frappe_contextTrue需要工作线程因此 handler 必须是普通函数——如果对async def打上该标志注册时就会直接抛出TypeError对应测试test_async_handler_with_frappe_context_rejected。在 dispatch.py 的_run_handlers中还有两道运行时闸门安装作用域闸门handler.app not in socket.installed_apps时跳过访客闸门socket.user Guest且allow_guestFalse时跳过。安装作用域重要一个 handler 只有在其所属应用安装于当前连接的站点时才会运行。所属应用在导入时自动从模块归属中检测Registry.importing_app上下文管理器打标签核心 handler 归属frappe见 registry.py 与 handlers.py你无需显式声明。因此your_app/realtime/handlers.py中的 handler 只会为安装了your_app的站点上的 socket 运行不会跨应用、跨站点泄漏。这一行为由测试test_install_scoping_skips_uninstalled_app与test_install_scoping_runs_installed_app直接验证。Socket 对象Socket是传给 handler 的类型化封装socket.py它薄薄地包装了 python-socketio 的AsyncServer绑定单个 sid namespace和连接时鉴权得到的Session。只读身份信息连接时从 Web 进程填充socket.site # str —— 该 socket 所在的站点 socket.user # str —— 匿名用户为 Guest socket.user_type # str —— 例如 System User socket.installed_apps # list[str]房间与发射await socket.join(room) # 把当前 socket 加入房间 await socket.leave(room) # 移出房间 await socket.emit(event, dataNone, roomNone) # 向房间发射room 为 None 时发给当前客户端权限检查默认、廉价——实时进程不做 DB 操作await socket.has_permission(doctype, nameNone) - bool # 通过 HTTP 回叫 Web 进程has_permission内部委托给Session.has_permissionauth.py请求 Web 进程的/api/method/frappe.realtime.has_permission接口。由于它是纯异步的不会占用工作线程也就不会排在阻塞型 handler 后面等待。在普通defhandler 中同样的调用去掉await即可SyncSocket.has_permission会把协程提交回事件循环。瞬时的逐 socket 状态断开连接时清空await socket.set(key, value) socket.get(key, defaultNone)这些状态存放在Session.data字典中并通过sio.save_session持久化普通 handler 中去掉await即可。核心 handler 就利用它跟踪subscribed_documents列表来实现谁在看这个文档的在线状态广播见 handlers.py 的doc_open/doc_close/notify_doc_viewers。权限检查的两种方式每个 handler 应二选一不要混用1. HTTP默认、推荐await socket.has_permission(doctype, name)与核心 handler 完全一致回问 Web 进程实时进程内不建立数据库连接代价低。没有特殊理由时都该用它。2. 进程内frappe_contextTrue为 handler 体打开一个完整的 Frappe 上下文可以直接调用frappe.has_permission(...)、查询数据库等realtime.on(project_subscribe, frappe_contextTrue) def project_subscribe(socket: Socket, project: str) - None: if frappe.has_permission(Project, docproject, ptyperead): socket.join(fproject:{project})代价是每一条此类事件都要完整走一遍frappe.init - connect - set_user - commit/rollback - destroy的周期并强制在实时进程内建立数据库连接。务必谨慎使用。其实现位于 context.pyfrappe.init(site, forceTrue)forceTrue是必须的因为frappe.local是 ContextVar 背后共享的可变字典只有init(forceTrue)会为该次调用重新绑定一份随后connect、set_user异常时rollback最终destroy。由于函数体运行在工作线程中常规的阻塞式 DB 驱动没有问题。从 Web 进程向客户端推送事件Web 进程不直接向 socket 发射事件而是通过 Redis 发布由实时服务器桥接给已连接的 socket。publish_realtime本身没有变化frappe.realtime模块提供了一组命名的辅助封装见init.pyfrom frappe.realtime import publish_to_room publish_to_room(project:PROJ-0001, project_updated, {status: Open})你发布时使用的房间字符串必须与 handler 中join的房间完全一致事件才能到达对应客户端。全部辅助函数每个辅助函数都是publish_realtime的薄封装——线上行为一致只是按目标房间命名。它们都接受关键字专属参数*, after_commitFalse推迟到当前事务提交后再发布publish_to_user(user, event, messageNone, *, after_commitFalse) publish_to_doc(doctype, docname, event, messageNone, *, after_commitFalse) publish_to_doctype(doctype, event, messageNone, *, after_commitFalse) publish_task_progress(task_id, messageNone, *, after_commitFalse) # 注意没有 event 参数 publish_to_website(event, messageNone, *, after_commitFalse) publish_to_all(event, messageNone, *, after_commitFalse) publish_to_room(room, event, messageNone, *, after_commitFalse)房间映射辅助函数房间可达范围publish_to_useruser:{user}该用户在本站点上的所有 socketpublish_to_docdoc:{doctype}/{docname}订阅了该文档的 socketpublish_to_doctypedoctype:{doctype}订阅了该 DocType 的 socketpublish_task_progresstask_progress:{task_id}正在观察该任务触发task_progress的 socketpublish_to_websitewebsite本站点上的每一个 socketpublish_to_allall本站点的 System User不含 Guestpublish_to_room{room}你传入的任意房间字符串关于 after_commit 的底层实现publish_realtime在after_commitTrue时不会立即通过 Redis 发射而是把(event, message, room)追加到frappe.local._realtime_log并注册frappe.db.after_commit/frappe.db.after_rollback回调flush_realtime_log/clear_realtime_log。事务提交后统一走emit_via_redis发射回滚则清空日志。这让事务内的实时通知与数据提交保持一致。注意task_id分支会强制after_commitFalse并自动把task_id写入 message。publish_to_all 的站点作用域publish_to_all是站点作用域的all房间等于本站点的 System User。它不是构建事件build events所用的那种跨站点、无房间广播——后者在publish_realtime内部走roomNone的分支bridge 中无 room 消息会向所有已连接的站点命名空间广播该路径始终保留给内部使用。规则永远不要在async defhandler 中阻塞事件循环。阻塞式 I/O 和密集 CPU 循环会拖垮所有其他 socket。如果工作会阻塞就把 handler 写成普通def让它在工作线程中运行。房间名和事件名是与浏览器客户端共享的契约要保持稳定一旦变更需同时更新socketio_client.js一类的客户端代码否则两端会失配。参考核心源码与测试索引如果你希望深入理解本指南背后的实现细节可以从以下文件继续服务器主体frappe/realtime/server.pyRealtimeServer、serve、main注册中心与realtime.on装饰器frappe/realtime/registry.py类型化 Socket 与阻塞式 SyncSocketfrappe/realtime/socket.py事件分发、安装作用域与访客闸门frappe/realtime/dispatch.py连接鉴权与 HTTP 权限回叫frappe/realtime/auth.pyRedis 到 Socket.IO 的桥接frappe/realtime/bridge.py配置解析frappe/realtime/config.py内置核心 handlerconnect、doc_subscribe、doc_open等frappe/realtime/handlers.py发布辅助函数与publish_realtimefrappe/realtime/init.py浏览器端客户端frappe/public/js/frappe/socketio_client.js单元测试覆盖鉴权、注册、作用域、Socket、分发、桥接、核心 handler、发布辅助函数frappe/tests/test_realtime_py.py【免费下载链接】frappeLow code web framework for real world applications, in Python and Javascript项目地址: https://gitcode.com/GitHub_Trending/fr/frappe创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考