update to use latest jupyter_events

Author: Zsailer

jupyter_server/base/handlers.py

@@ -20,6 +20,7 @@
 import prometheus_client
 from jinja2 import TemplateNotFound
 from jupyter_core.paths import is_hidden
+from jupyter_events import EventLogger
 from tornado import web
 from tornado.log import app_log
 from traitlets.config import Application
@@ -335,7 +336,7 @@ def config_manager(self):
         return self.settings["config_manager"]
 
     @property
-    def event_logger(self):
+    def event_logger(self) -> EventLogger:
         return self.settings["event_logger"]
 
     # ---------------------------------------------------------------

jupyter_server/serverapp.py

@@ -1951,6 +1951,11 @@ def init_logging(self):
     def init_event_logger(self):
         """Initialize the Event Bus."""
         self.event_logger = EventLogger(parent=self)
+        # event_schema_paths = [
+        #     ""
+        # ]
+        # for schema_path in event_schema_paths:
+        #     self.event_logger.register_event_schema(schema_path)
 
     def init_webapp(self):
         """initialize tornado webapp"""

jupyter_server/services/events/handlers.py

@@ -2,10 +2,11 @@
 
 .. versionadded:: 2.0
 """
-import logging
+import json
 from datetime import datetime
 from typing import Any, Dict, Optional
 
+from jupyter_events import EventLogger
 from tornado import web, websocket
 
 from jupyter_server.auth import authorized
@@ -16,18 +17,6 @@
 AUTH_RESOURCE = "events"
 
 
-class WebSocketLoggingHandler(logging.Handler):
-    """Python logging handler that routes records to a Tornado websocket."""
-
-    def __init__(self, websocket, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.websocket = websocket
-
-    def emit(self, record):
-        """Emit the message across the websocket"""
-        self.websocket.write_message(record.msg)
-
-
 class SubscribeWebsocket(
     JupyterHandler,
     websocket.WebSocketHandler,
@@ -56,16 +45,18 @@ async def get(self, *args, **kwargs):
         res = super().get(*args, **kwargs)
         await res
 
+    async def event_listener(self, logger: EventLogger, schema_id: str, data: dict) -> None:
+        capsule = dict(schema_id=schema_id, **data)
+        self.write_message(json.dumps(capsule))
+
     def open(self):
         """Routes events that are emitted by Jupyter Server's
         EventBus to a WebSocket client in the browser.
         """
-        self.logging_handler = WebSocketLoggingHandler(self)
-        # To do: add an eventlog.add_handler method to jupyter_events.
-        self.event_logger.register_handler(self.logging_handler)
+        self.event_logger.add_listener(listener=self.event_listener)
 
     def on_close(self):
-        self.event_logger.remove_handler(self.logging_handler)
+        self.event_logger.remove_listener(listener=self.event_listener)
 
 
 def validate_model(data: Dict[str, Any]) -> None:
@@ -110,7 +101,6 @@ async def post(self):
             validate_model(payload)
             self.event_logger.emit(
                 schema_id=payload.get("schema_id"),
-                version=payload.get("version"),
                 data=payload.get("data"),
                 timestamp_override=get_timestamp(payload),
             )

tests/services/events/mockextension/mock_extension.py

@@ -9,7 +9,6 @@ def get(self):
         # Emit an event.
         self.event_logger.emit(
             schema_id="event.mockextension.jupyter.org/message",
-            version=1,
             data={"event_message": "Hello world, from mock extension!"},
         )
 

tests/services/events/test_api.py

@@ -27,14 +27,14 @@ def event_logger(event_logger_sink):
     return event_logger
 
 
-async def test_subscribe_websocket(jp_ws_fetch, event_logger):
+async def test_subscribe_websocket(event_logger, jp_ws_fetch):
     ws = await jp_ws_fetch("/api/events/subscribe")
 
     event_logger.emit(
         schema_id="event.mock.jupyter.org/message",
-        version=1,
         data={"event_message": "Hello, world!"},
     )
+    # await event_logger.gather_listeners()
     message = await ws.read_message()
     event_data = json.loads(message)
     ws.close()