Jupyter Events
Jupyter Events is an event system library that enables Jupyter Python applications (e.g., Jupyter Server, JupyterLab Server, JupyterHub) and extensions to emit structured data describing internal happenings. Other software can then listen and respond to these events. The current version is 0.12.0, with releases occurring periodically to introduce new features, improvements, and bug fixes.
Warnings
- breaking Applications previously using `jupyter-telemetry` (e.g., JupyterHub) must migrate to `jupyter-events`. The API for event emission and schema definition is different and direct compatibility is not maintained.
- gotcha Event schemas must be explicitly registered with the `EventLogger` using `register_event_schema()` before emitting events. If an event is emitted for an unregistered schema ID, it will not be recorded and may raise an error or be silently dropped depending on the `EventLogger`'s configuration or version.
- gotcha Sensitive data in event payloads requires careful handling, especially if event data is routed to client applications or persistent storage. While `jupyter-events` allows for redaction policies, an incorrect or missing policy could expose sensitive information.
Install
-
pip install jupyter_events -
conda install -c conda-forge jupyter_events
Imports
- EventLogger
from jupyter_events import EventLogger
Quickstart
import logging
from jupyter_events import EventLogger
import json
import os
# Define a simple event schema (in-memory for this example)
my_event_schema = {
"$id": "https://example.com/schemas/my_event.json",
"title": "My Custom Event",
"description": "A simple custom event for demonstration.",
"type": "object",
"properties": {
"message": {"type": "string"},
"level": {"type": "string", "enum": ["info", "warning", "error"]}
},
"required": ["message", "level"]
}
# Create a temporary log file
event_log_file = 'quickstart_events.log'
# Initialize EventLogger with a FileHandler and register the schema
logger = EventLogger(
handlers=[logging.FileHandler(event_log_file)],
allowed_schemas=[my_event_schema["$id"]]
)
logger.register_event_schema(my_event_schema)
# Emit an event
print(f"Emitting an 'info' event...")
logger.emit(schema_id=my_event_schema["$id"], data={'message': 'This is a test event!', 'level': 'info'})
print(f"Event emitted to {event_log_file}")
# Read the logged event
with open(event_log_file, 'r') as f:
logged_event = json.loads(f.readline())
print(f"Logged event: {logged_event}")
# Clean up the log file
os.remove(event_log_file)