Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get(self, format, path):
exporter = get_exporter(format, config=self.config, log=self.log)
path = path.strip('/')
# If the notebook relates to a real file (default contents manager),
# give its path to nbconvert.
if hasattr(self.contents_manager, '_get_os_path'):
os_path = self.contents_manager._get_os_path(path)
ext_resources_dir, basename = os.path.split(os_path)
else:
ext_resources_dir = None
model = yield maybe_future(self.contents_manager.get(path=path))
name = model['name']
if model['type'] != 'notebook':
# not a notebook, redirect to files
return FilesRedirectHandler.redirect_to_files(self, path)
nb = model['content']
self.set_header('Last-Modified', model['last_modified'])
# create resources dictionary
mod_date = model['last_modified'].strftime(text.date_format)
nb_title = os.path.splitext(name)[0]
resource_dict = {
"metadata": {
"name": nb_title,
PUT is very similar to POST, but the requester specifies the name,
whereas with POST, the server picks the name.
PUT /api/contents/path/Name.ipynb
Save notebook at ``path/Name.ipynb``. Notebook structure is specified
in `content` key of JSON request body. If content is not specified,
create a new empty notebook.
"""
model = self.get_json_body()
if model:
if model.get('copy_from'):
raise web.HTTPError(400, "Cannot copy with PUT, only POST")
exists = yield maybe_future(self.contents_manager.file_exists(path))
if exists:
yield maybe_future(self._save(model, path))
else:
yield maybe_future(self._upload(model, path))
else:
yield maybe_future(self._new_untitled(path))
def _upload(self, model, path):
"""Handle upload of a new file to path"""
self.log.info(u"Uploading file to %s", path)
model = yield maybe_future(self.contents_manager.new(model, path))
self.set_status(201)
validate_model(model, expect_content=False)
self._finish_model(model)
def put(self, path=''):
"""Saves the file in the location specified by name and path.
PUT is very similar to POST, but the requester specifies the name,
whereas with POST, the server picks the name.
PUT /api/contents/path/Name.ipynb
Save notebook at ``path/Name.ipynb``. Notebook structure is specified
in `content` key of JSON request body. If content is not specified,
create a new empty notebook.
"""
model = self.get_json_body()
if model:
if model.get('copy_from'):
raise web.HTTPError(400, "Cannot copy with PUT, only POST")
exists = yield maybe_future(self.contents_manager.file_exists(path))
if exists:
yield maybe_future(self._save(model, path))
else:
yield maybe_future(self._upload(model, path))
else:
yield maybe_future(self._new_untitled(path))
mtype = model['type']
except KeyError:
raise web.HTTPError(400, "Missing field in JSON data: type")
name = model.get('name', None)
kernel = model.get('kernel', {})
kernel_name = kernel.get('name', None)
kernel_id = kernel.get('id', None)
if not kernel_id and not kernel_name:
self.log.debug("No kernel specified, using default kernel")
kernel_name = None
exists = yield maybe_future(sm.session_exists(path=path))
if exists:
model = yield maybe_future(sm.get_session(path=path))
else:
try:
model = yield maybe_future(
sm.create_session(path=path, kernel_name=kernel_name,
kernel_id=kernel_id, name=name,
type=mtype))
except NoSuchKernel:
msg = ("The '%s' kernel is not available. Please pick another "
"suitable kernel instead, or install that kernel." % kernel_name)
status_msg = '%s not found' % kernel_name
self.log.warning('Kernel not found: %s' % kernel_name)
self.set_status(501)
self.finish(json.dumps(dict(message=msg, short_message=status_msg)))
return
location = url_path_join(self.base_url, 'api', 'sessions', model['id'])
raise web.HTTPError(400, "No such kernel: %s" % kernel_id)
changes['kernel_id'] = kernel_id
elif model['kernel'].get('name') is not None:
kernel_name = model['kernel']['name']
kernel_id = yield sm.start_kernel_for_session(
session_id, kernel_name=kernel_name, name=before['name'],
path=before['path'], type=before['type'])
changes['kernel_id'] = kernel_id
yield maybe_future(sm.update_session(session_id, **changes))
model = yield maybe_future(sm.get_session(session_id=session_id))
if model['kernel']['id'] != before['kernel']['id']:
# kernel_id changed because we got a new kernel
# shutdown the old one
yield maybe_future(
km.shutdown_kernel(before['kernel']['id'])
)
self.finish(json.dumps(model, default=date_default))
def post(self, kernel_id, action):
km = self.kernel_manager
if action == 'interrupt':
km.interrupt_kernel(kernel_id)
self.set_status(204)
if action == 'restart':
try:
yield maybe_future(km.restart_kernel(kernel_id))
except Exception as e:
self.log.error("Exception restarting kernel", exc_info=True)
self.set_status(500)
else:
model = yield maybe_future(km.kernel_model(kernel_id))
self.write(json.dumps(model, default=date_default))
self.finish()
name = model.get('name', None)
kernel = model.get('kernel', {})
kernel_name = kernel.get('name', None)
kernel_id = kernel.get('id', None)
if not kernel_id and not kernel_name:
self.log.debug("No kernel specified, using default kernel")
kernel_name = None
exists = yield maybe_future(sm.session_exists(path=path))
if exists:
model = yield maybe_future(sm.get_session(path=path))
else:
try:
model = yield maybe_future(
sm.create_session(path=path, kernel_name=kernel_name,
kernel_id=kernel_id, name=name,
type=mtype))
except NoSuchKernel:
msg = ("The '%s' kernel is not available. Please pick another "
"suitable kernel instead, or install that kernel." % kernel_name)
status_msg = '%s not found' % kernel_name
self.log.warning('Kernel not found: %s' % kernel_name)
self.set_status(501)
self.finish(json.dumps(dict(message=msg, short_message=status_msg)))
return
location = url_path_join(self.base_url, 'api', 'sessions', model['id'])
self.set_header('Location', location)
self.set_status(201)
self.finish(json.dumps(model, default=date_default))
def delete(self, kernel_id):
km = self.kernel_manager
yield maybe_future(km.shutdown_kernel(kernel_id))
self.set_status(204)
self.finish()
def get(self):
# Return a list of running sessions
sm = self.session_manager
sessions = yield maybe_future(sm.list_sessions())
self.finish(json.dumps(sessions, default=date_default))