Merge pull request #4656 from takluyver/nbconvert-service
Nbconvert HTTP service
commit
556627a637
@ -0,0 +1,117 @@
|
||||
import io
|
||||
import os
|
||||
import zipfile
|
||||
|
||||
from tornado import web
|
||||
|
||||
from ..base.handlers import IPythonHandler, notebook_path_regex
|
||||
from IPython.nbformat.current import to_notebook_json
|
||||
from IPython.nbconvert.exporters.export import exporter_map
|
||||
from IPython.utils import tz
|
||||
from IPython.utils.py3compat import cast_bytes
|
||||
|
||||
import sys
|
||||
|
||||
def find_resource_files(output_files_dir):
|
||||
files = []
|
||||
for dirpath, dirnames, filenames in os.walk(output_files_dir):
|
||||
files.extend([os.path.join(dirpath, f) for f in filenames])
|
||||
return files
|
||||
|
||||
def respond_zip(handler, name, output, resources):
|
||||
"""Zip up the output and resource files and respond with the zip file.
|
||||
|
||||
Returns True if it has served a zip file, False if there are no resource
|
||||
files, in which case we serve the plain output file.
|
||||
"""
|
||||
# Check if we have resource files we need to zip
|
||||
output_files = resources.get('outputs', None)
|
||||
if not output_files:
|
||||
return False
|
||||
|
||||
# Headers
|
||||
zip_filename = os.path.splitext(name)[0] + '.zip'
|
||||
handler.set_header('Content-Disposition',
|
||||
'attachment; filename="%s"' % zip_filename)
|
||||
handler.set_header('Content-Type', 'application/zip')
|
||||
|
||||
# Prepare the zip file
|
||||
buffer = io.BytesIO()
|
||||
zipf = zipfile.ZipFile(buffer, mode='w', compression=zipfile.ZIP_DEFLATED)
|
||||
output_filename = os.path.splitext(name)[0] + '.' + resources['output_extension']
|
||||
zipf.writestr(output_filename, cast_bytes(output, 'utf-8'))
|
||||
for filename, data in output_files.items():
|
||||
zipf.writestr(os.path.basename(filename), data)
|
||||
zipf.close()
|
||||
|
||||
handler.finish(buffer.getvalue())
|
||||
return True
|
||||
|
||||
class NbconvertFileHandler(IPythonHandler):
|
||||
|
||||
SUPPORTED_METHODS = ('GET',)
|
||||
|
||||
@web.authenticated
|
||||
def get(self, format, path='', name=None):
|
||||
exporter = exporter_map[format](config=self.config)
|
||||
|
||||
path = path.strip('/')
|
||||
os_path = self.notebook_manager.get_os_path(name, path)
|
||||
if not os.path.isfile(os_path):
|
||||
raise web.HTTPError(404, u'Notebook does not exist: %s' % name)
|
||||
|
||||
info = os.stat(os_path)
|
||||
self.set_header('Last-Modified', tz.utcfromtimestamp(info.st_mtime))
|
||||
|
||||
output, resources = exporter.from_filename(os_path)
|
||||
|
||||
if respond_zip(self, name, output, resources):
|
||||
return
|
||||
|
||||
# Force download if requested
|
||||
if self.get_argument('download', 'false').lower() == 'true':
|
||||
filename = os.path.splitext(name)[0] + '.' + resources['output_extension']
|
||||
self.set_header('Content-Disposition',
|
||||
'attachment; filename="%s"' % filename)
|
||||
|
||||
# MIME type
|
||||
if exporter.output_mimetype:
|
||||
self.set_header('Content-Type',
|
||||
'%s; charset=utf-8' % exporter.output_mimetype)
|
||||
|
||||
self.finish(output)
|
||||
|
||||
class NbconvertPostHandler(IPythonHandler):
|
||||
SUPPORTED_METHODS = ('POST',)
|
||||
|
||||
@web.authenticated
|
||||
def post(self, format):
|
||||
exporter = exporter_map[format](config=self.config)
|
||||
|
||||
model = self.get_json_body()
|
||||
nbnode = to_notebook_json(model['content'])
|
||||
|
||||
output, resources = exporter.from_notebook_node(nbnode)
|
||||
|
||||
if respond_zip(self, nbnode.metadata.name, output, resources):
|
||||
return
|
||||
|
||||
# MIME type
|
||||
if exporter.output_mimetype:
|
||||
self.set_header('Content-Type',
|
||||
'%s; charset=utf-8' % exporter.output_mimetype)
|
||||
|
||||
self.finish(output)
|
||||
|
||||
#-----------------------------------------------------------------------------
|
||||
# URL to handler mappings
|
||||
#-----------------------------------------------------------------------------
|
||||
|
||||
_format_regex = r"(?P<format>\w+)"
|
||||
|
||||
|
||||
default_handlers = [
|
||||
(r"/nbconvert/%s%s" % (_format_regex, notebook_path_regex),
|
||||
NbconvertFileHandler),
|
||||
(r"/nbconvert/%s" % _format_regex, NbconvertPostHandler),
|
||||
]
|
||||
@ -0,0 +1,120 @@
|
||||
# coding: utf-8
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from os.path import join as pjoin
|
||||
import shutil
|
||||
|
||||
import requests
|
||||
|
||||
from IPython.html.utils import url_path_join
|
||||
from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
|
||||
from IPython.nbformat.current import (new_notebook, write, new_worksheet,
|
||||
new_heading_cell, new_code_cell,
|
||||
new_output)
|
||||
|
||||
class NbconvertAPI(object):
|
||||
"""Wrapper for nbconvert API calls."""
|
||||
def __init__(self, base_url):
|
||||
self.base_url = base_url
|
||||
|
||||
def _req(self, verb, path, body=None, params=None):
|
||||
response = requests.request(verb,
|
||||
url_path_join(self.base_url, 'nbconvert', path),
|
||||
data=body, params=params,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def from_file(self, format, path, name, download=False):
|
||||
return self._req('GET', url_path_join(format, path, name),
|
||||
params={'download':download})
|
||||
|
||||
def from_post(self, format, nbmodel):
|
||||
body = json.dumps(nbmodel)
|
||||
return self._req('POST', format, body)
|
||||
|
||||
def list_formats(self):
|
||||
return self._req('GET', '')
|
||||
|
||||
png_green_pixel = base64.encodestring(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00'
|
||||
b'\x00\x00\x01\x00\x00x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDAT'
|
||||
b'\x08\xd7c\x90\xfb\xcf\x00\x00\x02\\\x01\x1e.~d\x87\x00\x00\x00\x00IEND\xaeB`\x82')
|
||||
|
||||
class APITest(NotebookTestBase):
|
||||
def setUp(self):
|
||||
nbdir = self.notebook_dir.name
|
||||
|
||||
if not os.path.isdir(pjoin(nbdir, 'foo')):
|
||||
os.mkdir(pjoin(nbdir, 'foo'))
|
||||
|
||||
nb = new_notebook(name='testnb')
|
||||
|
||||
ws = new_worksheet()
|
||||
nb.worksheets = [ws]
|
||||
ws.cells.append(new_heading_cell(u'Created by test ³'))
|
||||
cc1 = new_code_cell(input=u'print(2*6)')
|
||||
cc1.outputs.append(new_output(output_text=u'12'))
|
||||
cc1.outputs.append(new_output(output_png=png_green_pixel, output_type='pyout'))
|
||||
ws.cells.append(cc1)
|
||||
|
||||
with io.open(pjoin(nbdir, 'foo', 'testnb.ipynb'), 'w',
|
||||
encoding='utf-8') as f:
|
||||
write(nb, f, format='ipynb')
|
||||
|
||||
self.nbconvert_api = NbconvertAPI(self.base_url())
|
||||
|
||||
def tearDown(self):
|
||||
nbdir = self.notebook_dir.name
|
||||
|
||||
for dname in ['foo']:
|
||||
shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
|
||||
|
||||
def test_from_file(self):
|
||||
r = self.nbconvert_api.from_file('html', 'foo', 'testnb.ipynb')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn(u'text/html', r.headers['Content-Type'])
|
||||
self.assertIn(u'Created by test', r.text)
|
||||
self.assertIn(u'print', r.text)
|
||||
|
||||
r = self.nbconvert_api.from_file('python', 'foo', 'testnb.ipynb')
|
||||
self.assertIn(u'text/x-python', r.headers['Content-Type'])
|
||||
self.assertIn(u'print(2*6)', r.text)
|
||||
|
||||
def test_from_file_404(self):
|
||||
with assert_http_error(404):
|
||||
self.nbconvert_api.from_file('html', 'foo', 'thisdoesntexist.ipynb')
|
||||
|
||||
def test_from_file_download(self):
|
||||
r = self.nbconvert_api.from_file('python', 'foo', 'testnb.ipynb', download=True)
|
||||
content_disposition = r.headers['Content-Disposition']
|
||||
self.assertIn('attachment', content_disposition)
|
||||
self.assertIn('testnb.py', content_disposition)
|
||||
|
||||
def test_from_file_zip(self):
|
||||
r = self.nbconvert_api.from_file('latex', 'foo', 'testnb.ipynb', download=True)
|
||||
self.assertIn(u'application/zip', r.headers['Content-Type'])
|
||||
self.assertIn(u'.zip', r.headers['Content-Disposition'])
|
||||
|
||||
def test_from_post(self):
|
||||
nbmodel_url = url_path_join(self.base_url(), 'api/notebooks/foo/testnb.ipynb')
|
||||
nbmodel = requests.get(nbmodel_url).json()
|
||||
|
||||
r = self.nbconvert_api.from_post(format='html', nbmodel=nbmodel)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn(u'text/html', r.headers['Content-Type'])
|
||||
self.assertIn(u'Created by test', r.text)
|
||||
self.assertIn(u'print', r.text)
|
||||
|
||||
r = self.nbconvert_api.from_post(format='python', nbmodel=nbmodel)
|
||||
self.assertIn(u'text/x-python', r.headers['Content-Type'])
|
||||
self.assertIn(u'print(2*6)', r.text)
|
||||
|
||||
def test_from_post_zip(self):
|
||||
nbmodel_url = url_path_join(self.base_url(), 'api/notebooks/foo/testnb.ipynb')
|
||||
nbmodel = requests.get(nbmodel_url).json()
|
||||
|
||||
r = self.nbconvert_api.from_post(format='latex', nbmodel=nbmodel)
|
||||
self.assertIn(u'application/zip', r.headers['Content-Type'])
|
||||
self.assertIn(u'.zip', r.headers['Content-Disposition'])
|
||||
@ -0,0 +1,23 @@
|
||||
import json
|
||||
|
||||
from tornado import web
|
||||
|
||||
from ...base.handlers import IPythonHandler, json_errors
|
||||
from IPython.nbconvert.exporters.export import exporter_map
|
||||
|
||||
class NbconvertRootHandler(IPythonHandler):
|
||||
SUPPORTED_METHODS = ('GET',)
|
||||
|
||||
@web.authenticated
|
||||
@json_errors
|
||||
def get(self):
|
||||
res = {}
|
||||
for format, exporter in exporter_map.items():
|
||||
res[format] = info = {}
|
||||
info['output_mimetype'] = exporter.output_mimetype
|
||||
|
||||
self.finish(json.dumps(res))
|
||||
|
||||
default_handlers = [
|
||||
(r"/api/nbconvert", NbconvertRootHandler),
|
||||
]
|
||||
@ -0,0 +1,31 @@
|
||||
import requests
|
||||
|
||||
from IPython.html.utils import url_path_join
|
||||
from IPython.html.tests.launchnotebook import NotebookTestBase
|
||||
|
||||
class NbconvertAPI(object):
|
||||
"""Wrapper for nbconvert API calls."""
|
||||
def __init__(self, base_url):
|
||||
self.base_url = base_url
|
||||
|
||||
def _req(self, verb, path, body=None, params=None):
|
||||
response = requests.request(verb,
|
||||
url_path_join(self.base_url, 'api/nbconvert', path),
|
||||
data=body, params=params,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def list_formats(self):
|
||||
return self._req('GET', '')
|
||||
|
||||
class APITest(NotebookTestBase):
|
||||
def setUp(self):
|
||||
self.nbconvert_api = NbconvertAPI(self.base_url())
|
||||
|
||||
def test_list_formats(self):
|
||||
formats = self.nbconvert_api.list_formats().json()
|
||||
self.assertIsInstance(formats, dict)
|
||||
self.assertIn('python', formats)
|
||||
self.assertIn('html', formats)
|
||||
self.assertEqual(formats['python']['output_mimetype'], 'text/x-python')
|
||||
Loading…
Reference in new issue