Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def load_jupyter_server_extension(nbapp):
"""
Called during notebook start
"""
api_url= url_path_join(nbapp.web_app.settings['base_url'], '/api/nbresuse/v1')
nbapp.web_app.add_handlers('.*', [
(api_url, MetricsHandler, {'nbapp': nbapp})
])
"""
Holds server-side configuration for nbresuse
"""
process_memory_metrics = List(
trait=PSUtilMetric(),
default_value=[{"name": "memory_info", "attribute": "rss"}],
)
system_memory_metrics = List(
trait=PSUtilMetric(),
default_value=[{"name": "virtual_memory", "attribute": "total"}],
)
process_cpu_metrics = List(
trait=PSUtilMetric(),
default_value=[{"name": "cpu_percent", "kwargs": {"interval": 0.05}}],
)
system_cpu_metrics = List(
trait=PSUtilMetric(), default_value=[{"name": "cpu_count"}]
)
mem_warning_threshold = Float(
default_value=0.1,
help="""
Warn user with flashing lights when memory usage is within this fraction
memory limit.
For example, if memory limit is 128MB, `mem_warning_threshold` is 0.1,
we will start warning the user when they use (128 - (128 * 0.1)) MB.
import os
import json
from notebook.utils import url_path_join
from notebook.base.handlers import IPythonHandler
from tornado import web, iostream
import asyncio
import pluggy
from nbresuse import hooks, default_resources
from collections import ChainMap
plugin_manager = pluggy.PluginManager('nbresuse')
plugin_manager.add_hookspecs(hooks)
# Register the resources nbresuse provides by default
plugin_manager.register(default_resources)
class MetricsHandler(IPythonHandler):
def initialize(self, nbapp):
self.set_header('content-type', 'text/event-stream')
self.set_header('cache-control', 'no-cache')
self.nbapp = nbapp
@web.authenticated
async def get(self):
"""
Calculate and return current resource usage metrics
"""
while True:
metrics = {}
for metric_response in plugin_manager.hook.nbresuse_add_resource(nbapp=self.nbapp):
def setup_handlers(web_app):
route_pattern = url_path_join(web_app.settings['base_url'], '/metrics')
web_app.add_handlers('.*', [(route_pattern, MetricsHandler)])
import os
import json
from notebook.utils import url_path_join
from notebook.base.handlers import IPythonHandler
from tornado import web, iostream
import asyncio
import pluggy
from nbresuse import hooks, default_resources
from collections import ChainMap
plugin_manager = pluggy.PluginManager('nbresuse')
plugin_manager.add_hookspecs(hooks)
# Register the resources nbresuse provides by default
plugin_manager.register(default_resources)
class MetricsHandler(IPythonHandler):
def initialize(self, nbapp):
self.set_header('content-type', 'text/event-stream')
self.set_header('cache-control', 'no-cache')
self.nbapp = nbapp
@web.authenticated
async def get(self):
"""
Calculate and return current resource usage metrics
"""
while True: