update to jupyter_events 0.2.0

Author: Zsailer

jupyter_server/base/handlers.py

@@ -335,8 +335,8 @@ def config_manager(self):
         return self.settings["config_manager"]
 
     @property
-    def event_bus(self):
-        return self.settings["event_bus"]
+    def event_logger(self):
+        return self.settings["event_logger"]
 
     # ---------------------------------------------------------------
     # CORS

jupyter_server/serverapp.py

@@ -61,6 +61,7 @@
 from jupyter_client.session import Session
 from jupyter_core.application import JupyterApp, base_aliases, base_flags
 from jupyter_core.paths import jupyter_runtime_dir
+from jupyter_events.logger import EventLogger
 from nbformat.sign import NotebookNotary
 from traitlets import (
     Any,
@@ -123,7 +124,6 @@
     AsyncContentsManager,
     ContentsManager,
 )
-from jupyter_server.services.events.bus import EventBus
 from jupyter_server.services.kernels.kernelmanager import (
     AsyncMappingKernelManager,
     MappingKernelManager,
@@ -212,7 +212,7 @@ def __init__(
         session_manager,
         kernel_spec_manager,
         config_manager,
-        event_bus,
+        event_logger,
         extra_services,
         log,
         base_url,
@@ -248,7 +248,7 @@ def __init__(
             session_manager,
             kernel_spec_manager,
             config_manager,
-            event_bus,
+            event_logger,
             extra_services,
             log,
             base_url,
@@ -270,7 +270,7 @@ def init_settings(
         session_manager,
         kernel_spec_manager,
         config_manager,
-        event_bus,
+        event_logger,
         extra_services,
         log,
         base_url,
@@ -359,7 +359,7 @@ def init_settings(
             config_manager=config_manager,
             authorizer=authorizer,
             identity_provider=identity_provider,
-            event_bus=event_bus,
+            event_logger=event_logger,
             # handlers
             extra_services=extra_services,
             # Jupyter stuff
@@ -770,7 +770,7 @@ class ServerApp(JupyterApp):
         GatewaySessionManager,
         GatewayClient,
         Authorizer,
-        EventBus,
+        EventLogger,
     ]
 
     subcommands = dict(
@@ -1552,10 +1552,10 @@ def _default_kernel_spec_manager_class(self):
         ),
     )
 
-    event_bus = Instance(
-        EventBus,
+    event_logger = Instance(
+        EventLogger,
         allow_none=True,
-        help="An EventBus for emitting structured event data from Jupyter Server and extensions.",
+        help="An EventLogger for emitting structured event data from Jupyter Server and extensions.",
     )
 
     info_file = Unicode()
@@ -1948,9 +1948,9 @@ def init_logging(self):
         logger.parent = self.log
         logger.setLevel(self.log.level)
 
-    def init_eventbus(self):
+    def init_event_logger(self):
         """Initialize the Event Bus."""
-        self.event_bus = EventBus.instance(parent=self)
+        self.event_logger = EventLogger(parent=self)
 
     def init_webapp(self):
         """initialize tornado webapp"""
@@ -2012,7 +2012,7 @@ def init_webapp(self):
             self.session_manager,
             self.kernel_spec_manager,
             self.config_manager,
-            self.event_bus,
+            self.event_logger,
             self.extra_services,
             self.log,
             self.base_url,
@@ -2487,7 +2487,7 @@ def initialize(
         if find_extensions:
             self.find_server_extensions()
         self.init_logging()
-        self.init_eventbus()
+        self.init_event_logger()
         self.init_server_extensions()
 
         # Special case the starter extension and load
@@ -2814,8 +2814,6 @@ async def _cleanup(self):
         await self.cleanup_kernels()
         if getattr(self, "session_manager", None):
             self.session_manager.close()
-        if getattr(self, "event_bus", None):
-            self.event_bus.clear_instance()
 
     def start_ioloop(self):
         """Start the IO Loop."""

jupyter_server/services/events/bus.py

@@ -6,7 +6,6 @@
 from datetime import datetime
 from typing import Any, Dict, Optional
 
-from jupyter_events.logger import _skip_message
 from pythonjsonlogger import jsonlogger
 from tornado import web, websocket
 
@@ -63,21 +62,16 @@ def open(self):
         EventBus to a WebSocket client in the browser.
         """
         self.logging_handler = WebSocketLoggingHandler(self)
-        # Add a JSON formatter to the handler.
-        formatter = jsonlogger.JsonFormatter(json_serializer=_skip_message)
-        self.logging_handler.setFormatter(formatter)
         # To do: add an eventlog.add_handler method to jupyter_events.
-        self.event_bus.log.addHandler(self.logging_handler)
-        self.event_bus.handlers.append(self.logging_handler)
+        self.event_logger.register_handler(self.logging_handler)
 
     def on_close(self):
-        self.event_bus.log.removeHandler(self.logging_handler)
-        self.event_bus.handlers.remove(self.logging_handler)
+        self.event_logger.remove_handler(self.logging_handler)
 
 
 def validate_model(data: Dict[str, Any]) -> None:
     """Validates for required fields in the JSON request body"""
-    required_keys = {"schema_name", "version", "event"}
+    required_keys = {"schema_id", "version", "event"}
     for key in required_keys:
         if key not in data:
             raise web.HTTPError(400, f"Missing `{key}` in the JSON request body.")
@@ -115,10 +109,10 @@ async def post(self):
 
         try:
             validate_model(payload)
-            self.event_bus.record_event(
-                schema_name=payload.get("schema_name"),
+            self.event_logger.emit(
+                schema_id=payload.get("schema_id"),
                 version=payload.get("version"),
-                event=payload.get("event"),
+                data=payload.get("event"),
                 timestamp_override=get_timestamp(payload),
             )
             self.set_status(204)

jupyter_server/services/events/handlers.py

@@ -44,7 +44,7 @@ dependencies = [
     "tornado>=6.1.0",
     "traitlets>=5.1",
     "websocket-client",
-    "jupyter_events>=0.1.0"
+    "jupyter_events>=0.2.0"
 ]
 
 [project.urls]

pyproject.toml

@@ -7,17 +7,17 @@
 class MockEventHandler(JupyterHandler):
     def get(self):
         # Emit an event.
-        self.event_bus.record_event(
-            schema_name="event.mockextension.jupyter.org/message",
+        self.event_logger.emit(
+            schema_id="event.mockextension.jupyter.org/message",
             version=1,
-            event={"event_message": "Hello world, from mock extension!"},
+            data={"event_message": "Hello world, from mock extension!"},
         )
 
 
 def _load_jupyter_server_extension(serverapp):
     # Register a schema with the EventBus
     schema_file = pathlib.Path(__file__).parent / "mock_extension_event.yaml"
-    serverapp.event_bus.register_schema_file(schema_file)
+    serverapp.event_logger.register_event_schema(schema_file)
     serverapp.web_app.add_handlers(
         ".*$", [(url_path_join(serverapp.base_url, "/mock/event"), MockEventHandler)]
     )

tests/services/events/mockextension/mock_extension.py

@@ -5,42 +5,36 @@
 
 import pytest
 import tornado
-from jupyter_events.logger import _skip_message
 from pythonjsonlogger import jsonlogger
 
 from tests.utils import expected_http_error
 
 
 @pytest.fixture
-def eventbus_sink(jp_serverapp):
-    event_bus = jp_serverapp.event_bus
+def event_logger_sink(jp_serverapp):
+    event_logger = jp_serverapp.event_logger
     # Register the event schema defined in this directory.
     schema_file = pathlib.Path(__file__).parent / "mock_event.yaml"
-    event_bus.register_schema_file(schema_file)
-    event_bus.allowed_schemas = ["event.mock.jupyter.org/message"]
-    event_bus.allowed_categories = ["category.jupyter.org/unrestricted"]
+    event_logger.register_event_schema(schema_file)
     sink = io.StringIO()
-    formatter = jsonlogger.JsonFormatter(json_serializer=_skip_message)
     handler = logging.StreamHandler(sink)
-    handler.setFormatter(formatter)
-    event_bus.handlers = [handler]
-    event_bus.log.addHandler(handler)
-    return event_bus, sink
+    event_logger.register_handler(handler)
+    return event_logger, sink
 
 
 @pytest.fixture
-def event_bus(eventbus_sink):
-    event_bus, sink = eventbus_sink
-    return event_bus
+def event_logger(event_logger_sink):
+    event_logger, sink = event_logger_sink
+    return event_logger
 
 
-async def test_subscribe_websocket(jp_ws_fetch, event_bus):
+async def test_subscribe_websocket(jp_ws_fetch, event_logger):
     ws = await jp_ws_fetch("/api/events/subscribe")
 
-    event_bus.record_event(
-        schema_name="event.mock.jupyter.org/message",
+    event_logger.emit(
+        schema_id="event.mock.jupyter.org/message",
         version=1,
-        event={"event_message": "Hello, world!"},
+        data={"event_message": "Hello, world!"},
     )
     message = await ws.read_message()
     event_data = json.loads(message)
@@ -51,7 +45,7 @@ async def test_subscribe_websocket(jp_ws_fetch, event_bus):
 
 payload_1 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 1,
     "event": {
         "event_message": "Hello, world!"
@@ -62,7 +56,7 @@ async def test_subscribe_websocket(jp_ws_fetch, event_bus):
 
 payload_2 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 1,
     "event": {
         "event_message": "Hello, world!"
@@ -72,8 +66,8 @@ async def test_subscribe_websocket(jp_ws_fetch, event_bus):
 
 
 @pytest.mark.parametrize("payload", [payload_1, payload_2])
-async def test_post_event(jp_fetch, eventbus_sink, payload):
-    event_bus, sink = eventbus_sink
+async def test_post_event(jp_fetch, event_logger_sink, payload):
+    event_logger, sink = event_logger_sink
 
     r = await jp_fetch("api", "events", method="POST", body=payload)
     assert r.code == 204
@@ -90,7 +84,7 @@ async def test_post_event(jp_fetch, eventbus_sink, payload):
 
 payload_3 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "event": {
         "event_message": "Hello, world!"
     }
@@ -108,14 +102,14 @@ async def test_post_event(jp_fetch, eventbus_sink, payload):
 
 payload_5 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 1
 }
 """
 
 payload_6 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 1,
     "event": {
         "event_message": "Hello, world!"
@@ -126,7 +120,7 @@ async def test_post_event(jp_fetch, eventbus_sink, payload):
 
 
 @pytest.mark.parametrize("payload", [payload_3, payload_4, payload_5, payload_6])
-async def test_post_event_400(jp_fetch, event_bus, payload):
+async def test_post_event_400(jp_fetch, event_logger, payload):
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
         await jp_fetch("api", "events", method="POST", body=payload)
 
@@ -135,7 +129,7 @@ async def test_post_event_400(jp_fetch, event_bus, payload):
 
 payload_7 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 1,
     "event": {
         "message": "Hello, world!"
@@ -145,7 +139,7 @@ async def test_post_event_400(jp_fetch, event_bus, payload):
 
 payload_8 = """\
 {
-    "schema_name": "event.mock.jupyter.org/message",
+    "schema_id": "event.mock.jupyter.org/message",
     "version": 2,
     "event": {
         "message": "Hello, world!"
@@ -155,7 +149,7 @@ async def test_post_event_400(jp_fetch, event_bus, payload):
 
 
 @pytest.mark.parametrize("payload", [payload_7, payload_8])
-async def test_post_event_500(jp_fetch, event_bus, payload):
+async def test_post_event_500(jp_fetch, event_logger, payload):
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
         await jp_fetch("api", "events", method="POST", body=payload)