##// END OF EJS Templates
Addressing review comments....
Brian E. Granger -
Show More
@@ -1,459 +1,457 b''
1 1 """A notebook manager that uses the local file system for storage.
2 2
3 3 Authors:
4 4
5 5 * Brian Granger
6 6 * Zach Sailer
7 7 """
8 8
9 9 #-----------------------------------------------------------------------------
10 10 # Copyright (C) 2011 The IPython Development Team
11 11 #
12 12 # Distributed under the terms of the BSD License. The full license is in
13 13 # the file COPYING, distributed as part of this software.
14 14 #-----------------------------------------------------------------------------
15 15
16 16 #-----------------------------------------------------------------------------
17 17 # Imports
18 18 #-----------------------------------------------------------------------------
19 19
20 20 import io
21 21 import itertools
22 22 import os
23 23 import glob
24 24 import shutil
25 25
26 26 from tornado import web
27 27
28 28 from .nbmanager import NotebookManager
29 29 from IPython.nbformat import current
30 30 from IPython.utils.traitlets import Unicode, Dict, Bool, TraitError
31 31 from IPython.utils import tz
32 32
33 33 #-----------------------------------------------------------------------------
34 34 # Classes
35 35 #-----------------------------------------------------------------------------
36 36
37 37 class FileNotebookManager(NotebookManager):
38 38
39 39 save_script = Bool(False, config=True,
40 40 help="""Automatically create a Python script when saving the notebook.
41 41
42 42 For easier use of import, %run and %load across notebooks, a
43 43 <notebook-name>.py script will be created next to any
44 44 <notebook-name>.ipynb on each save. This can also be set with the
45 45 short `--script` flag.
46 46 """
47 47 )
48 48
49 49 checkpoint_dir = Unicode(config=True,
50 50 help="""The location in which to keep notebook checkpoints
51 51
52 52 By default, it is notebook-dir/.ipynb_checkpoints
53 53 """
54 54 )
55 55 def _checkpoint_dir_default(self):
56 56 return os.path.join(self.notebook_dir, '.ipynb_checkpoints')
57 57
58 58 def _checkpoint_dir_changed(self, name, old, new):
59 59 """do a bit of validation of the checkpoint dir"""
60 60 if not os.path.isabs(new):
61 61 # If we receive a non-absolute path, make it absolute.
62 62 abs_new = os.path.abspath(new)
63 63 self.checkpoint_dir = abs_new
64 64 return
65 65 if os.path.exists(new) and not os.path.isdir(new):
66 66 raise TraitError("checkpoint dir %r is not a directory" % new)
67 67 if not os.path.exists(new):
68 68 self.log.info("Creating checkpoint dir %s", new)
69 69 try:
70 70 os.mkdir(new)
71 71 except:
72 72 raise TraitError("Couldn't create checkpoint dir %r" % new)
73 73
74 74 def get_notebook_names(self, path=''):
75 75 """List all notebook names in the notebook dir and path."""
76 76 path = path.strip('/')
77 77 if not os.path.isdir(self.get_os_path(path=path)):
78 78 raise web.HTTPError(404, 'Directory not found: ' + path)
79 79 names = glob.glob(self.get_os_path('*'+self.filename_ext, path))
80 80 names = [os.path.basename(name)
81 81 for name in names]
82 82 return names
83 83
84 84 def increment_filename(self, basename, path='', ext='.ipynb'):
85 85 """Return a non-used filename of the form basename<int>."""
86 86 path = path.strip('/')
87 87 for i in itertools.count():
88 88 name = u'{basename}{i}{ext}'.format(basename=basename, i=i, ext=ext)
89 89 os_path = self.get_os_path(name, path)
90 90 if not os.path.isfile(os_path):
91 91 break
92 92 return name
93 93
94 94 def path_exists(self, path):
95 95 """Does the API-style path (directory) actually exist?
96 96
97 97 Parameters
98 98 ----------
99 99 path : string
100 100 The path to check. This is an API path (`/` separated,
101 101 relative to base notebook-dir).
102 102
103 103 Returns
104 104 -------
105 105 exists : bool
106 106 Whether the path is indeed a directory.
107 107 """
108 108 path = path.strip('/')
109 109 os_path = self.get_os_path(path=path)
110 110 return os.path.isdir(os_path)
111 111
112 112 def get_os_path(self, name=None, path=''):
113 113 """Given a notebook name and a URL path, return its file system
114 114 path.
115 115
116 116 Parameters
117 117 ----------
118 118 name : string
119 119 The name of a notebook file with the .ipynb extension
120 120 path : string
121 121 The relative URL path (with '/' as separator) to the named
122 122 notebook.
123 123
124 124 Returns
125 125 -------
126 126 path : string
127 127 A file system path that combines notebook_dir (location where
128 128 server started), the relative path, and the filename with the
129 129 current operating system's url.
130 130 """
131 131 parts = path.strip('/').split('/')
132 132 parts = [p for p in parts if p != ''] # remove duplicate splits
133 133 if name is not None:
134 134 parts.append(name)
135 135 path = os.path.join(self.notebook_dir, *parts)
136 136 return path
137 137
138 138 def notebook_exists(self, name, path=''):
139 139 """Returns a True if the notebook exists. Else, returns False.
140 140
141 141 Parameters
142 142 ----------
143 143 name : string
144 144 The name of the notebook you are checking.
145 145 path : string
146 146 The relative path to the notebook (with '/' as separator)
147 147
148 148 Returns
149 149 -------
150 150 bool
151 151 """
152 152 path = path.strip('/')
153 153 nbpath = self.get_os_path(name, path=path)
154 154 return os.path.isfile(nbpath)
155 155
156 156 # TODO: Remove this after we create the contents web service and directories are
157 157 # no longer listed by the notebook web service.
158 158 def list_dirs(self, path):
159 159 """List the directories for a given API style path."""
160 160 path = path.strip('/')
161 161 os_path = self.get_os_path('', path)
162 if not os.path.isdir(os_path):
163 raise web.HTTPError(404, u'diretory does not exist: %r' % os_path)
162 164 dir_names = os.listdir(os_path)
163 165 dirs = []
164 166 for name in dir_names:
165 167 os_path = self.get_os_path(name, path)
166 168 if os.path.isdir(os_path) and not name.startswith('.'):
169 try:
167 170 model = self.get_dir_model(name, path)
171 except IOError:
172 pass
168 173 dirs.append(model)
169 174 dirs = sorted(dirs, key=lambda item: item['name'])
170 175 return dirs
171 176
172 177 # TODO: Remove this after we create the contents web service and directories are
173 178 # no longer listed by the notebook web service.
174 179 def get_dir_model(self, name, path=''):
175 180 """Get the directory model given a directory name and its API style path"""
176 181 path = path.strip('/')
177 182 os_path = self.get_os_path(name, path)
178 183 if not os.path.isdir(os_path):
179 184 raise IOError('directory does not exist: %r' % os_path)
180 185 info = os.stat(os_path)
181 186 last_modified = tz.utcfromtimestamp(info.st_mtime)
182 187 created = tz.utcfromtimestamp(info.st_ctime)
183 188 # Create the notebook model.
184 189 model ={}
185 190 model['name'] = name
186 191 model['path'] = path
187 192 model['last_modified'] = last_modified
188 193 model['created'] = created
189 194 model['type'] = 'directory'
190 195 return model
191 196
192 197 def list_notebooks(self, path):
193 198 """Returns a list of dictionaries that are the standard model
194 199 for all notebooks in the relative 'path'.
195 200
196 201 Parameters
197 202 ----------
198 203 path : str
199 204 the URL path that describes the relative path for the
200 205 listed notebooks
201 206
202 207 Returns
203 208 -------
204 209 notebooks : list of dicts
205 210 a list of the notebook models without 'content'
206 211 """
207 212 path = path.strip('/')
208 213 notebook_names = self.get_notebook_names(path)
209 index = []
210 notebooks = []
211 for name in notebook_names:
212 model = self.get_notebook_model(name, path, content=False)
213 if name.lower() == 'index.ipynb':
214 index.append(model)
215 else:
216 notebooks.append(model)
214 notebooks = [self.get_notebook_model(name, path, content=False) for name in notebook_names]
217 215 notebooks = sorted(notebooks, key=lambda item: item['name'])
218 notebooks = index + self.list_dirs(path) + notebooks
219 216 return notebooks
220 217
221 218 def get_notebook_model(self, name, path='', content=True):
222 219 """ Takes a path and name for a notebook and returns its model
223 220
224 221 Parameters
225 222 ----------
226 223 name : str
227 224 the name of the notebook
228 225 path : str
229 226 the URL path that describes the relative path for
230 227 the notebook
231 228
232 229 Returns
233 230 -------
234 231 model : dict
235 232 the notebook model. If contents=True, returns the 'contents'
236 233 dict in the model as well.
237 234 """
238 235 path = path.strip('/')
239 236 if not self.notebook_exists(name=name, path=path):
240 237 raise web.HTTPError(404, u'Notebook does not exist: %s' % name)
241 238 os_path = self.get_os_path(name, path)
242 239 info = os.stat(os_path)
243 240 last_modified = tz.utcfromtimestamp(info.st_mtime)
244 241 created = tz.utcfromtimestamp(info.st_ctime)
245 242 # Create the notebook model.
246 243 model ={}
247 244 model['name'] = name
248 245 model['path'] = path
249 246 model['last_modified'] = last_modified
250 247 model['created'] = created
248 model['type'] = 'notebook'
251 249 if content:
252 250 with io.open(os_path, 'r', encoding='utf-8') as f:
253 251 try:
254 252 nb = current.read(f, u'json')
255 253 except Exception as e:
256 254 raise web.HTTPError(400, u"Unreadable Notebook: %s %s" % (os_path, e))
257 255 self.mark_trusted_cells(nb, path, name)
258 256 model['content'] = nb
259 257 return model
260 258
261 259 def save_notebook_model(self, model, name='', path=''):
262 260 """Save the notebook model and return the model with no content."""
263 261 path = path.strip('/')
264 262
265 263 if 'content' not in model:
266 264 raise web.HTTPError(400, u'No notebook JSON data provided')
267 265
268 266 # One checkpoint should always exist
269 267 if self.notebook_exists(name, path) and not self.list_checkpoints(name, path):
270 268 self.create_checkpoint(name, path)
271 269
272 270 new_path = model.get('path', path).strip('/')
273 271 new_name = model.get('name', name)
274 272
275 273 if path != new_path or name != new_name:
276 274 self.rename_notebook(name, path, new_name, new_path)
277 275
278 276 # Save the notebook file
279 277 os_path = self.get_os_path(new_name, new_path)
280 278 nb = current.to_notebook_json(model['content'])
281 279
282 280 self.check_and_sign(nb, new_path, new_name)
283 281
284 282 if 'name' in nb['metadata']:
285 283 nb['metadata']['name'] = u''
286 284 try:
287 285 self.log.debug("Autosaving notebook %s", os_path)
288 286 with io.open(os_path, 'w', encoding='utf-8') as f:
289 287 current.write(nb, f, u'json')
290 288 except Exception as e:
291 289 raise web.HTTPError(400, u'Unexpected error while autosaving notebook: %s %s' % (os_path, e))
292 290
293 291 # Save .py script as well
294 292 if self.save_script:
295 293 py_path = os.path.splitext(os_path)[0] + '.py'
296 294 self.log.debug("Writing script %s", py_path)
297 295 try:
298 296 with io.open(py_path, 'w', encoding='utf-8') as f:
299 297 current.write(nb, f, u'py')
300 298 except Exception as e:
301 299 raise web.HTTPError(400, u'Unexpected error while saving notebook as script: %s %s' % (py_path, e))
302 300
303 301 model = self.get_notebook_model(new_name, new_path, content=False)
304 302 return model
305 303
306 304 def update_notebook_model(self, model, name, path=''):
307 305 """Update the notebook's path and/or name"""
308 306 path = path.strip('/')
309 307 new_name = model.get('name', name)
310 308 new_path = model.get('path', path).strip('/')
311 309 if path != new_path or name != new_name:
312 310 self.rename_notebook(name, path, new_name, new_path)
313 311 model = self.get_notebook_model(new_name, new_path, content=False)
314 312 return model
315 313
316 314 def delete_notebook_model(self, name, path=''):
317 315 """Delete notebook by name and path."""
318 316 path = path.strip('/')
319 317 os_path = self.get_os_path(name, path)
320 318 if not os.path.isfile(os_path):
321 319 raise web.HTTPError(404, u'Notebook does not exist: %s' % os_path)
322 320
323 321 # clear checkpoints
324 322 for checkpoint in self.list_checkpoints(name, path):
325 323 checkpoint_id = checkpoint['id']
326 324 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
327 325 if os.path.isfile(cp_path):
328 326 self.log.debug("Unlinking checkpoint %s", cp_path)
329 327 os.unlink(cp_path)
330 328
331 329 self.log.debug("Unlinking notebook %s", os_path)
332 330 os.unlink(os_path)
333 331
334 332 def rename_notebook(self, old_name, old_path, new_name, new_path):
335 333 """Rename a notebook."""
336 334 old_path = old_path.strip('/')
337 335 new_path = new_path.strip('/')
338 336 if new_name == old_name and new_path == old_path:
339 337 return
340 338
341 339 new_os_path = self.get_os_path(new_name, new_path)
342 340 old_os_path = self.get_os_path(old_name, old_path)
343 341
344 342 # Should we proceed with the move?
345 343 if os.path.isfile(new_os_path):
346 344 raise web.HTTPError(409, u'Notebook with name already exists: %s' % new_os_path)
347 345 if self.save_script:
348 346 old_py_path = os.path.splitext(old_os_path)[0] + '.py'
349 347 new_py_path = os.path.splitext(new_os_path)[0] + '.py'
350 348 if os.path.isfile(new_py_path):
351 349 raise web.HTTPError(409, u'Python script with name already exists: %s' % new_py_path)
352 350
353 351 # Move the notebook file
354 352 try:
355 353 os.rename(old_os_path, new_os_path)
356 354 except Exception as e:
357 355 raise web.HTTPError(500, u'Unknown error renaming notebook: %s %s' % (old_os_path, e))
358 356
359 357 # Move the checkpoints
360 358 old_checkpoints = self.list_checkpoints(old_name, old_path)
361 359 for cp in old_checkpoints:
362 360 checkpoint_id = cp['id']
363 361 old_cp_path = self.get_checkpoint_path(checkpoint_id, old_name, old_path)
364 362 new_cp_path = self.get_checkpoint_path(checkpoint_id, new_name, new_path)
365 363 if os.path.isfile(old_cp_path):
366 364 self.log.debug("Renaming checkpoint %s -> %s", old_cp_path, new_cp_path)
367 365 os.rename(old_cp_path, new_cp_path)
368 366
369 367 # Move the .py script
370 368 if self.save_script:
371 369 os.rename(old_py_path, new_py_path)
372 370
373 371 # Checkpoint-related utilities
374 372
375 373 def get_checkpoint_path(self, checkpoint_id, name, path=''):
376 374 """find the path to a checkpoint"""
377 375 path = path.strip('/')
378 376 basename, _ = os.path.splitext(name)
379 377 filename = u"{name}-{checkpoint_id}{ext}".format(
380 378 name=basename,
381 379 checkpoint_id=checkpoint_id,
382 380 ext=self.filename_ext,
383 381 )
384 382 cp_path = os.path.join(path, self.checkpoint_dir, filename)
385 383 return cp_path
386 384
387 385 def get_checkpoint_model(self, checkpoint_id, name, path=''):
388 386 """construct the info dict for a given checkpoint"""
389 387 path = path.strip('/')
390 388 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
391 389 stats = os.stat(cp_path)
392 390 last_modified = tz.utcfromtimestamp(stats.st_mtime)
393 391 info = dict(
394 392 id = checkpoint_id,
395 393 last_modified = last_modified,
396 394 )
397 395 return info
398 396
399 397 # public checkpoint API
400 398
401 399 def create_checkpoint(self, name, path=''):
402 400 """Create a checkpoint from the current state of a notebook"""
403 401 path = path.strip('/')
404 402 nb_path = self.get_os_path(name, path)
405 403 # only the one checkpoint ID:
406 404 checkpoint_id = u"checkpoint"
407 405 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
408 406 self.log.debug("creating checkpoint for notebook %s", name)
409 407 if not os.path.exists(self.checkpoint_dir):
410 408 os.mkdir(self.checkpoint_dir)
411 409 shutil.copy2(nb_path, cp_path)
412 410
413 411 # return the checkpoint info
414 412 return self.get_checkpoint_model(checkpoint_id, name, path)
415 413
416 414 def list_checkpoints(self, name, path=''):
417 415 """list the checkpoints for a given notebook
418 416
419 417 This notebook manager currently only supports one checkpoint per notebook.
420 418 """
421 419 path = path.strip('/')
422 420 checkpoint_id = "checkpoint"
423 421 path = self.get_checkpoint_path(checkpoint_id, name, path)
424 422 if not os.path.exists(path):
425 423 return []
426 424 else:
427 425 return [self.get_checkpoint_model(checkpoint_id, name, path)]
428 426
429 427
430 428 def restore_checkpoint(self, checkpoint_id, name, path=''):
431 429 """restore a notebook to a checkpointed state"""
432 430 path = path.strip('/')
433 431 self.log.info("restoring Notebook %s from checkpoint %s", name, checkpoint_id)
434 432 nb_path = self.get_os_path(name, path)
435 433 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
436 434 if not os.path.isfile(cp_path):
437 435 self.log.debug("checkpoint file does not exist: %s", cp_path)
438 436 raise web.HTTPError(404,
439 437 u'Notebook checkpoint does not exist: %s-%s' % (name, checkpoint_id)
440 438 )
441 439 # ensure notebook is readable (never restore from an unreadable notebook)
442 440 with io.open(cp_path, 'r', encoding='utf-8') as f:
443 441 nb = current.read(f, u'json')
444 442 shutil.copy2(cp_path, nb_path)
445 443 self.log.debug("copying %s -> %s", cp_path, nb_path)
446 444
447 445 def delete_checkpoint(self, checkpoint_id, name, path=''):
448 446 """delete a notebook's checkpoint"""
449 447 path = path.strip('/')
450 448 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
451 449 if not os.path.isfile(cp_path):
452 450 raise web.HTTPError(404,
453 451 u'Notebook checkpoint does not exist: %s%s-%s' % (path, name, checkpoint_id)
454 452 )
455 453 self.log.debug("unlinking %s", cp_path)
456 454 os.unlink(cp_path)
457 455
458 456 def info_string(self):
459 457 return "Serving notebooks from local directory: %s" % self.notebook_dir
@@ -1,280 +1,290 b''
1 1 """Tornado handlers for the notebooks web service.
2 2
3 3 Authors:
4 4
5 5 * Brian Granger
6 6 """
7 7
8 8 #-----------------------------------------------------------------------------
9 9 # Copyright (C) 2011 The IPython Development Team
10 10 #
11 11 # Distributed under the terms of the BSD License. The full license is in
12 12 # the file COPYING, distributed as part of this software.
13 13 #-----------------------------------------------------------------------------
14 14
15 15 #-----------------------------------------------------------------------------
16 16 # Imports
17 17 #-----------------------------------------------------------------------------
18 18
19 19 import json
20 20
21 21 from tornado import web
22 22
23 23 from IPython.html.utils import url_path_join, url_escape
24 24 from IPython.utils.jsonutil import date_default
25 25
26 26 from IPython.html.base.handlers import (IPythonHandler, json_errors,
27 27 notebook_path_regex, path_regex,
28 28 notebook_name_regex)
29 29
30 30 #-----------------------------------------------------------------------------
31 31 # Notebook web service handlers
32 32 #-----------------------------------------------------------------------------
33 33
34 34
35 35 class NotebookHandler(IPythonHandler):
36 36
37 37 SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
38 38
39 39 def notebook_location(self, name, path=''):
40 40 """Return the full URL location of a notebook based.
41 41
42 42 Parameters
43 43 ----------
44 44 name : unicode
45 45 The base name of the notebook, such as "foo.ipynb".
46 46 path : unicode
47 47 The URL path of the notebook.
48 48 """
49 49 return url_escape(url_path_join(
50 50 self.base_project_url, 'api', 'notebooks', path, name
51 51 ))
52 52
53 53 def _finish_model(self, model, location=True):
54 54 """Finish a JSON request with a model, setting relevant headers, etc."""
55 55 if location:
56 56 location = self.notebook_location(model['name'], model['path'])
57 57 self.set_header('Location', location)
58 58 self.set_header('Last-Modified', model['last_modified'])
59 59 self.finish(json.dumps(model, default=date_default))
60 60
61 61 @web.authenticated
62 62 @json_errors
63 63 def get(self, path='', name=None):
64 64 """Return a Notebook or list of notebooks.
65 65
66 66 * GET with path and no notebook name lists notebooks in a directory
67 67 * GET with path and notebook name returns notebook JSON
68 68 """
69 69 nbm = self.notebook_manager
70 70 # Check to see if a notebook name was given
71 71 if name is None:
72 # List notebooks in 'path'
73 notebooks = nbm.list_notebooks(path)
72 # TODO: Remove this after we create the contents web service and directories are
73 # no longer listed by the notebook web service. This should only handle notebooks
74 # and not directories.
75 dirs = nbm.list_dirs(path)
76 notebooks = []
77 index = []
78 for nb in nbm.list_notebooks(path):
79 if nb['name'].lower() == 'index.ipynb':
80 index.append(nb)
81 else:
82 notebooks.append(nb)
83 notebooks = index + dirs + notebooks
74 84 self.finish(json.dumps(notebooks, default=date_default))
75 85 return
76 86 # get and return notebook representation
77 87 model = nbm.get_notebook_model(name, path)
78 88 self._finish_model(model, location=False)
79 89
80 90 @web.authenticated
81 91 @json_errors
82 92 def patch(self, path='', name=None):
83 93 """PATCH renames a notebook without re-uploading content."""
84 94 nbm = self.notebook_manager
85 95 if name is None:
86 96 raise web.HTTPError(400, u'Notebook name missing')
87 97 model = self.get_json_body()
88 98 if model is None:
89 99 raise web.HTTPError(400, u'JSON body missing')
90 100 model = nbm.update_notebook_model(model, name, path)
91 101 self._finish_model(model)
92 102
93 103 def _copy_notebook(self, copy_from, path, copy_to=None):
94 104 """Copy a notebook in path, optionally specifying the new name.
95 105
96 106 Only support copying within the same directory.
97 107 """
98 108 self.log.info(u"Copying notebook from %s/%s to %s/%s",
99 109 path, copy_from,
100 110 path, copy_to or '',
101 111 )
102 112 model = self.notebook_manager.copy_notebook(copy_from, copy_to, path)
103 113 self.set_status(201)
104 114 self._finish_model(model)
105 115
106 116 def _upload_notebook(self, model, path, name=None):
107 117 """Upload a notebook
108 118
109 119 If name specified, create it in path/name.
110 120 """
111 121 self.log.info(u"Uploading notebook to %s/%s", path, name or '')
112 122 if name:
113 123 model['name'] = name
114 124
115 125 model = self.notebook_manager.create_notebook_model(model, path)
116 126 self.set_status(201)
117 127 self._finish_model(model)
118 128
119 129 def _create_empty_notebook(self, path, name=None):
120 130 """Create an empty notebook in path
121 131
122 132 If name specified, create it in path/name.
123 133 """
124 134 self.log.info(u"Creating new notebook in %s/%s", path, name or '')
125 135 model = {}
126 136 if name:
127 137 model['name'] = name
128 138 model = self.notebook_manager.create_notebook_model(model, path=path)
129 139 self.set_status(201)
130 140 self._finish_model(model)
131 141
132 142 def _save_notebook(self, model, path, name):
133 143 """Save an existing notebook."""
134 144 self.log.info(u"Saving notebook at %s/%s", path, name)
135 145 model = self.notebook_manager.save_notebook_model(model, name, path)
136 146 if model['path'] != path.strip('/') or model['name'] != name:
137 147 # a rename happened, set Location header
138 148 location = True
139 149 else:
140 150 location = False
141 151 self._finish_model(model, location)
142 152
143 153 @web.authenticated
144 154 @json_errors
145 155 def post(self, path='', name=None):
146 156 """Create a new notebook in the specified path.
147 157
148 158 POST creates new notebooks. The server always decides on the notebook name.
149 159
150 160 POST /api/notebooks/path
151 161 New untitled notebook in path. If content specified, upload a
152 162 notebook, otherwise start empty.
153 163 POST /api/notebooks/path?copy=OtherNotebook.ipynb
154 164 New copy of OtherNotebook in path
155 165 """
156 166
157 167 if name is not None:
158 168 raise web.HTTPError(400, "Only POST to directories. Use PUT for full names.")
159 169
160 170 model = self.get_json_body()
161 171
162 172 if model is not None:
163 173 copy_from = model.get('copy_from')
164 174 if copy_from:
165 175 if model.get('content'):
166 176 raise web.HTTPError(400, "Can't upload and copy at the same time.")
167 177 self._copy_notebook(copy_from, path)
168 178 else:
169 179 self._upload_notebook(model, path)
170 180 else:
171 181 self._create_empty_notebook(path)
172 182
173 183 @web.authenticated
174 184 @json_errors
175 185 def put(self, path='', name=None):
176 186 """Saves the notebook in the location specified by name and path.
177 187
178 188 PUT is very similar to POST, but the requester specifies the name,
179 189 whereas with POST, the server picks the name.
180 190
181 191 PUT /api/notebooks/path/Name.ipynb
182 192 Save notebook at ``path/Name.ipynb``. Notebook structure is specified
183 193 in `content` key of JSON request body. If content is not specified,
184 194 create a new empty notebook.
185 195 PUT /api/notebooks/path/Name.ipynb?copy=OtherNotebook.ipynb
186 196 Copy OtherNotebook to Name
187 197 """
188 198 if name is None:
189 199 raise web.HTTPError(400, "Only PUT to full names. Use POST for directories.")
190 200
191 201 model = self.get_json_body()
192 202 if model:
193 203 copy_from = model.get('copy_from')
194 204 if copy_from:
195 205 if model.get('content'):
196 206 raise web.HTTPError(400, "Can't upload and copy at the same time.")
197 207 self._copy_notebook(copy_from, path, name)
198 208 elif self.notebook_manager.notebook_exists(name, path):
199 209 self._save_notebook(model, path, name)
200 210 else:
201 211 self._upload_notebook(model, path, name)
202 212 else:
203 213 self._create_empty_notebook(path, name)
204 214
205 215 @web.authenticated
206 216 @json_errors
207 217 def delete(self, path='', name=None):
208 218 """delete the notebook in the given notebook path"""
209 219 nbm = self.notebook_manager
210 220 nbm.delete_notebook_model(name, path)
211 221 self.set_status(204)
212 222 self.finish()
213 223
214 224
215 225 class NotebookCheckpointsHandler(IPythonHandler):
216 226
217 227 SUPPORTED_METHODS = ('GET', 'POST')
218 228
219 229 @web.authenticated
220 230 @json_errors
221 231 def get(self, path='', name=None):
222 232 """get lists checkpoints for a notebook"""
223 233 nbm = self.notebook_manager
224 234 checkpoints = nbm.list_checkpoints(name, path)
225 235 data = json.dumps(checkpoints, default=date_default)
226 236 self.finish(data)
227 237
228 238 @web.authenticated
229 239 @json_errors
230 240 def post(self, path='', name=None):
231 241 """post creates a new checkpoint"""
232 242 nbm = self.notebook_manager
233 243 checkpoint = nbm.create_checkpoint(name, path)
234 244 data = json.dumps(checkpoint, default=date_default)
235 245 location = url_path_join(self.base_project_url, 'api/notebooks',
236 246 path, name, 'checkpoints', checkpoint['id'])
237 247 self.set_header('Location', url_escape(location))
238 248 self.set_status(201)
239 249 self.finish(data)
240 250
241 251
242 252 class ModifyNotebookCheckpointsHandler(IPythonHandler):
243 253
244 254 SUPPORTED_METHODS = ('POST', 'DELETE')
245 255
246 256 @web.authenticated
247 257 @json_errors
248 258 def post(self, path, name, checkpoint_id):
249 259 """post restores a notebook from a checkpoint"""
250 260 nbm = self.notebook_manager
251 261 nbm.restore_checkpoint(checkpoint_id, name, path)
252 262 self.set_status(204)
253 263 self.finish()
254 264
255 265 @web.authenticated
256 266 @json_errors
257 267 def delete(self, path, name, checkpoint_id):
258 268 """delete clears a checkpoint for a given notebook"""
259 269 nbm = self.notebook_manager
260 270 nbm.delete_checkpoint(checkpoint_id, name, path)
261 271 self.set_status(204)
262 272 self.finish()
263 273
264 274 #-----------------------------------------------------------------------------
265 275 # URL to handler mappings
266 276 #-----------------------------------------------------------------------------
267 277
268 278
269 279 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
270 280
271 281 default_handlers = [
272 282 (r"/api/notebooks%s/checkpoints" % notebook_path_regex, NotebookCheckpointsHandler),
273 283 (r"/api/notebooks%s/checkpoints/%s" % (notebook_path_regex, _checkpoint_id_regex),
274 284 ModifyNotebookCheckpointsHandler),
275 285 (r"/api/notebooks%s" % notebook_path_regex, NotebookHandler),
276 286 (r"/api/notebooks%s" % path_regex, NotebookHandler),
277 287 ]
278 288
279 289
280 290
@@ -1,198 +1,218 b''
1 1 """A base class notebook manager.
2 2
3 3 Authors:
4 4
5 5 * Brian Granger
6 6 * Zach Sailer
7 7 """
8 8
9 9 #-----------------------------------------------------------------------------
10 10 # Copyright (C) 2011 The IPython Development Team
11 11 #
12 12 # Distributed under the terms of the BSD License. The full license is in
13 13 # the file COPYING, distributed as part of this software.
14 14 #-----------------------------------------------------------------------------
15 15
16 16 #-----------------------------------------------------------------------------
17 17 # Imports
18 18 #-----------------------------------------------------------------------------
19 19
20 20 import os
21 21
22 22 from IPython.config.configurable import LoggingConfigurable
23 23 from IPython.nbformat import current, sign
24 24 from IPython.utils import py3compat
25 25 from IPython.utils.traitlets import Instance, Unicode, TraitError
26 26
27 27 #-----------------------------------------------------------------------------
28 28 # Classes
29 29 #-----------------------------------------------------------------------------
30 30
31 31 class NotebookManager(LoggingConfigurable):
32 32
33 33 # Todo:
34 34 # The notebook_dir attribute is used to mean a couple of different things:
35 35 # 1. Where the notebooks are stored if FileNotebookManager is used.
36 36 # 2. The cwd of the kernel for a project.
37 37 # Right now we use this attribute in a number of different places and
38 38 # we are going to have to disentangle all of this.
39 39 notebook_dir = Unicode(py3compat.getcwd(), config=True, help="""
40 40 The directory to use for notebooks.
41 41 """)
42 42
43 43 filename_ext = Unicode(u'.ipynb')
44 44
45 45 notary = Instance(sign.NotebookNotary)
46 46 def _notary_default(self):
47 47 return sign.NotebookNotary(parent=self)
48 48
49 49 def check_and_sign(self, nb, path, name):
50 50 """Check for trusted cells, and sign the notebook.
51 51
52 52 Called as a part of saving notebooks.
53 53 """
54 54 if self.notary.check_cells(nb):
55 55 self.notary.sign(nb)
56 56 else:
57 57 self.log.warn("Saving untrusted notebook %s/%s", path, name)
58 58
59 59 def mark_trusted_cells(self, nb, path, name):
60 60 """Mark cells as trusted if the notebook signature matches.
61 61
62 62 Called as a part of loading notebooks.
63 63 """
64 64 trusted = self.notary.check_signature(nb)
65 65 if not trusted:
66 66 self.log.warn("Notebook %s/%s is not trusted", path, name)
67 67 self.notary.mark_cells(nb, trusted)
68 68
69 69 def path_exists(self, path):
70 70 """Does the API-style path (directory) actually exist?
71 71
72 72 Override this method in subclasses.
73 73
74 74 Parameters
75 75 ----------
76 76 path : string
77 77 The
78 78
79 79 Returns
80 80 -------
81 81 exists : bool
82 82 Whether the path does indeed exist.
83 83 """
84 84 raise NotImplementedError
85 85
86 86 def _notebook_dir_changed(self, name, old, new):
87 87 """Do a bit of validation of the notebook dir."""
88 88 if not os.path.isabs(new):
89 89 # If we receive a non-absolute path, make it absolute.
90 90 self.notebook_dir = os.path.abspath(new)
91 91 return
92 92 if os.path.exists(new) and not os.path.isdir(new):
93 93 raise TraitError("notebook dir %r is not a directory" % new)
94 94 if not os.path.exists(new):
95 95 self.log.info("Creating notebook dir %s", new)
96 96 try:
97 97 os.mkdir(new)
98 98 except:
99 99 raise TraitError("Couldn't create notebook dir %r" % new)
100 100
101 101 # Main notebook API
102 102
103 103 def increment_filename(self, basename, path=''):
104 104 """Increment a notebook filename without the .ipynb to make it unique.
105 105
106 106 Parameters
107 107 ----------
108 108 basename : unicode
109 109 The name of a notebook without the ``.ipynb`` file extension.
110 110 path : unicode
111 111 The URL path of the notebooks directory
112 112 """
113 113 return basename
114 114
115 # TODO: Remove this after we create the contents web service and directories are
116 # no longer listed by the notebook web service.
117 def list_dirs(self, path):
118 """List the directory models for a given API style path."""
119 raise NotImplementedError('must be implemented in a subclass')
120
121 # TODO: Remove this after we create the contents web service and directories are
122 # no longer listed by the notebook web service.
123 def get_dir_model(self, name, path=''):
124 """Get the directory model given a directory name and its API style path.
125
126 The keys in the model should be:
127 * name
128 * path
129 * last_modified
130 * created
131 * type='directory'
132 """
133 raise NotImplementedError('must be implemented in a subclass')
134
115 135 def list_notebooks(self, path=''):
116 136 """Return a list of notebook dicts without content.
117 137
118 138 This returns a list of dicts, each of the form::
119 139
120 140 dict(notebook_id=notebook,name=name)
121 141
122 142 This list of dicts should be sorted by name::
123 143
124 144 data = sorted(data, key=lambda item: item['name'])
125 145 """
126 146 raise NotImplementedError('must be implemented in a subclass')
127 147
128 148 def get_notebook_model(self, name, path='', content=True):
129 149 """Get the notebook model with or without content."""
130 150 raise NotImplementedError('must be implemented in a subclass')
131 151
132 152 def save_notebook_model(self, model, name, path=''):
133 153 """Save the notebook model and return the model with no content."""
134 154 raise NotImplementedError('must be implemented in a subclass')
135 155
136 156 def update_notebook_model(self, model, name, path=''):
137 157 """Update the notebook model and return the model with no content."""
138 158 raise NotImplementedError('must be implemented in a subclass')
139 159
140 160 def delete_notebook_model(self, name, path=''):
141 161 """Delete notebook by name and path."""
142 162 raise NotImplementedError('must be implemented in a subclass')
143 163
144 164 def create_notebook_model(self, model=None, path=''):
145 165 """Create a new notebook and return its model with no content."""
146 166 path = path.strip('/')
147 167 if model is None:
148 168 model = {}
149 169 if 'content' not in model:
150 170 metadata = current.new_metadata(name=u'')
151 171 model['content'] = current.new_notebook(metadata=metadata)
152 172 if 'name' not in model:
153 173 model['name'] = self.increment_filename('Untitled', path)
154 174
155 175 model['path'] = path
156 176 model = self.save_notebook_model(model, model['name'], model['path'])
157 177 return model
158 178
159 179 def copy_notebook(self, from_name, to_name=None, path=''):
160 180 """Copy an existing notebook and return its new model.
161 181
162 182 If to_name not specified, increment `from_name-Copy#.ipynb`.
163 183 """
164 184 path = path.strip('/')
165 185 model = self.get_notebook_model(from_name, path)
166 186 if not to_name:
167 187 base = os.path.splitext(from_name)[0] + '-Copy'
168 188 to_name = self.increment_filename(base, path)
169 189 model['name'] = to_name
170 190 model = self.save_notebook_model(model, to_name, path)
171 191 return model
172 192
173 193 # Checkpoint-related
174 194
175 195 def create_checkpoint(self, name, path=''):
176 196 """Create a checkpoint of the current state of a notebook
177 197
178 198 Returns a checkpoint_id for the new checkpoint.
179 199 """
180 200 raise NotImplementedError("must be implemented in a subclass")
181 201
182 202 def list_checkpoints(self, name, path=''):
183 203 """Return a list of checkpoints for a given notebook"""
184 204 return []
185 205
186 206 def restore_checkpoint(self, checkpoint_id, name, path=''):
187 207 """Restore a notebook from one of its checkpoints"""
188 208 raise NotImplementedError("must be implemented in a subclass")
189 209
190 210 def delete_checkpoint(self, checkpoint_id, name, path=''):
191 211 """delete a checkpoint for a notebook"""
192 212 raise NotImplementedError("must be implemented in a subclass")
193 213
194 214 def log_info(self):
195 215 self.log.info(self.info_string())
196 216
197 217 def info_string(self):
198 218 return "Serving notebooks"
@@ -1,329 +1,329 b''
1 1 # coding: utf-8
2 2 """Test the notebooks webservice API."""
3 3
4 4 import io
5 5 import json
6 6 import os
7 7 import shutil
8 8 from unicodedata import normalize
9 9
10 10 pjoin = os.path.join
11 11
12 12 import requests
13 13
14 14 from IPython.html.utils import url_path_join, url_escape
15 15 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
16 16 from IPython.nbformat import current
17 17 from IPython.nbformat.current import (new_notebook, write, read, new_worksheet,
18 18 new_heading_cell, to_notebook_json)
19 19 from IPython.nbformat import v2
20 20 from IPython.utils import py3compat
21 21 from IPython.utils.data import uniq_stable
22 22
23 23
24 24 # TODO: Remove this after we create the contents web service and directories are
25 25 # no longer listed by the notebook web service.
26 26 def notebooks_only(nb_list):
27 return [nb for nb in nb_list if 'type' not in nb]
27 return [nb for nb in nb_list if nb['type']=='notebook']
28 28
29 29
30 30 class NBAPI(object):
31 31 """Wrapper for notebook API calls."""
32 32 def __init__(self, base_url):
33 33 self.base_url = base_url
34 34
35 35 def _req(self, verb, path, body=None):
36 36 response = requests.request(verb,
37 37 url_path_join(self.base_url, 'api/notebooks', path),
38 38 data=body,
39 39 )
40 40 response.raise_for_status()
41 41 return response
42 42
43 43 def list(self, path='/'):
44 44 return self._req('GET', path)
45 45
46 46 def read(self, name, path='/'):
47 47 return self._req('GET', url_path_join(path, name))
48 48
49 49 def create_untitled(self, path='/'):
50 50 return self._req('POST', path)
51 51
52 52 def upload_untitled(self, body, path='/'):
53 53 return self._req('POST', path, body)
54 54
55 55 def copy_untitled(self, copy_from, path='/'):
56 56 body = json.dumps({'copy_from':copy_from})
57 57 return self._req('POST', path, body)
58 58
59 59 def create(self, name, path='/'):
60 60 return self._req('PUT', url_path_join(path, name))
61 61
62 62 def upload(self, name, body, path='/'):
63 63 return self._req('PUT', url_path_join(path, name), body)
64 64
65 65 def copy(self, copy_from, copy_to, path='/'):
66 66 body = json.dumps({'copy_from':copy_from})
67 67 return self._req('PUT', url_path_join(path, copy_to), body)
68 68
69 69 def save(self, name, body, path='/'):
70 70 return self._req('PUT', url_path_join(path, name), body)
71 71
72 72 def delete(self, name, path='/'):
73 73 return self._req('DELETE', url_path_join(path, name))
74 74
75 75 def rename(self, name, path, new_name):
76 76 body = json.dumps({'name': new_name})
77 77 return self._req('PATCH', url_path_join(path, name), body)
78 78
79 79 def get_checkpoints(self, name, path):
80 80 return self._req('GET', url_path_join(path, name, 'checkpoints'))
81 81
82 82 def new_checkpoint(self, name, path):
83 83 return self._req('POST', url_path_join(path, name, 'checkpoints'))
84 84
85 85 def restore_checkpoint(self, name, path, checkpoint_id):
86 86 return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
87 87
88 88 def delete_checkpoint(self, name, path, checkpoint_id):
89 89 return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id))
90 90
91 91 class APITest(NotebookTestBase):
92 92 """Test the kernels web service API"""
93 93 dirs_nbs = [('', 'inroot'),
94 94 ('Directory with spaces in', 'inspace'),
95 95 (u'unicodΓ©', 'innonascii'),
96 96 ('foo', 'a'),
97 97 ('foo', 'b'),
98 98 ('foo', 'name with spaces'),
99 99 ('foo', u'unicodΓ©'),
100 100 ('foo/bar', 'baz'),
101 101 (u'Γ₯ b', u'Γ§ d')
102 102 ]
103 103
104 104 dirs = uniq_stable([d for (d,n) in dirs_nbs])
105 105 del dirs[0] # remove ''
106 106
107 107 def setUp(self):
108 108 nbdir = self.notebook_dir.name
109 109
110 110 for d in self.dirs:
111 111 d.replace('/', os.sep)
112 112 if not os.path.isdir(pjoin(nbdir, d)):
113 113 os.mkdir(pjoin(nbdir, d))
114 114
115 115 for d, name in self.dirs_nbs:
116 116 d = d.replace('/', os.sep)
117 117 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
118 118 encoding='utf-8') as f:
119 119 nb = new_notebook(name=name)
120 120 write(nb, f, format='ipynb')
121 121
122 122 self.nb_api = NBAPI(self.base_url())
123 123
124 124 def tearDown(self):
125 125 nbdir = self.notebook_dir.name
126 126
127 127 for dname in ['foo', 'Directory with spaces in', u'unicodΓ©', u'Γ₯ b']:
128 128 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
129 129
130 130 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
131 131 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
132 132
133 133 def test_list_notebooks(self):
134 134 nbs = notebooks_only(self.nb_api.list().json())
135 135 self.assertEqual(len(nbs), 1)
136 136 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
137 137
138 138 nbs = notebooks_only(self.nb_api.list('/Directory with spaces in/').json())
139 139 self.assertEqual(len(nbs), 1)
140 140 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
141 141
142 142 nbs = notebooks_only(self.nb_api.list(u'/unicodΓ©/').json())
143 143 self.assertEqual(len(nbs), 1)
144 144 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
145 145 self.assertEqual(nbs[0]['path'], u'unicodΓ©')
146 146
147 147 nbs = notebooks_only(self.nb_api.list('/foo/bar/').json())
148 148 self.assertEqual(len(nbs), 1)
149 149 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
150 150 self.assertEqual(nbs[0]['path'], 'foo/bar')
151 151
152 152 nbs = notebooks_only(self.nb_api.list('foo').json())
153 153 self.assertEqual(len(nbs), 4)
154 154 nbnames = { normalize('NFC', n['name']) for n in nbs }
155 155 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
156 156 expected = { normalize('NFC', name) for name in expected }
157 157 self.assertEqual(nbnames, expected)
158 158
159 159 def test_list_nonexistant_dir(self):
160 160 with assert_http_error(404):
161 161 self.nb_api.list('nonexistant')
162 162
163 163 def test_get_contents(self):
164 164 for d, name in self.dirs_nbs:
165 165 nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
166 166 self.assertEqual(nb['name'], u'%s.ipynb' % name)
167 167 self.assertIn('content', nb)
168 168 self.assertIn('metadata', nb['content'])
169 169 self.assertIsInstance(nb['content']['metadata'], dict)
170 170
171 171 # Name that doesn't exist - should be a 404
172 172 with assert_http_error(404):
173 173 self.nb_api.read('q.ipynb', 'foo')
174 174
175 175 def _check_nb_created(self, resp, name, path):
176 176 self.assertEqual(resp.status_code, 201)
177 177 location_header = py3compat.str_to_unicode(resp.headers['Location'])
178 178 self.assertEqual(location_header, url_escape(url_path_join(u'/api/notebooks', path, name)))
179 179 self.assertEqual(resp.json()['name'], name)
180 180 assert os.path.isfile(pjoin(
181 181 self.notebook_dir.name,
182 182 path.replace('/', os.sep),
183 183 name,
184 184 ))
185 185
186 186 def test_create_untitled(self):
187 187 resp = self.nb_api.create_untitled(path=u'Γ₯ b')
188 188 self._check_nb_created(resp, 'Untitled0.ipynb', u'Γ₯ b')
189 189
190 190 # Second time
191 191 resp = self.nb_api.create_untitled(path=u'Γ₯ b')
192 192 self._check_nb_created(resp, 'Untitled1.ipynb', u'Γ₯ b')
193 193
194 194 # And two directories down
195 195 resp = self.nb_api.create_untitled(path='foo/bar')
196 196 self._check_nb_created(resp, 'Untitled0.ipynb', 'foo/bar')
197 197
198 198 def test_upload_untitled(self):
199 199 nb = new_notebook(name='Upload test')
200 200 nbmodel = {'content': nb}
201 201 resp = self.nb_api.upload_untitled(path=u'Γ₯ b',
202 202 body=json.dumps(nbmodel))
203 203 self._check_nb_created(resp, 'Untitled0.ipynb', u'Γ₯ b')
204 204
205 205 def test_upload(self):
206 206 nb = new_notebook(name=u'ignored')
207 207 nbmodel = {'content': nb}
208 208 resp = self.nb_api.upload(u'Upload tΓ©st.ipynb', path=u'Γ₯ b',
209 209 body=json.dumps(nbmodel))
210 210 self._check_nb_created(resp, u'Upload tΓ©st.ipynb', u'Γ₯ b')
211 211
212 212 def test_upload_v2(self):
213 213 nb = v2.new_notebook()
214 214 ws = v2.new_worksheet()
215 215 nb.worksheets.append(ws)
216 216 ws.cells.append(v2.new_code_cell(input='print("hi")'))
217 217 nbmodel = {'content': nb}
218 218 resp = self.nb_api.upload(u'Upload tΓ©st.ipynb', path=u'Γ₯ b',
219 219 body=json.dumps(nbmodel))
220 220 self._check_nb_created(resp, u'Upload tΓ©st.ipynb', u'Γ₯ b')
221 221 resp = self.nb_api.read(u'Upload tΓ©st.ipynb', u'Γ₯ b')
222 222 data = resp.json()
223 223 self.assertEqual(data['content']['nbformat'], current.nbformat)
224 224 self.assertEqual(data['content']['orig_nbformat'], 2)
225 225
226 226 def test_copy_untitled(self):
227 227 resp = self.nb_api.copy_untitled(u'Γ§ d.ipynb', path=u'Γ₯ b')
228 228 self._check_nb_created(resp, u'Γ§ d-Copy0.ipynb', u'Γ₯ b')
229 229
230 230 def test_copy(self):
231 231 resp = self.nb_api.copy(u'Γ§ d.ipynb', u'cΓΈpy.ipynb', path=u'Γ₯ b')
232 232 self._check_nb_created(resp, u'cΓΈpy.ipynb', u'Γ₯ b')
233 233
234 234 def test_delete(self):
235 235 for d, name in self.dirs_nbs:
236 236 resp = self.nb_api.delete('%s.ipynb' % name, d)
237 237 self.assertEqual(resp.status_code, 204)
238 238
239 239 for d in self.dirs + ['/']:
240 240 nbs = notebooks_only(self.nb_api.list(d).json())
241 241 self.assertEqual(len(nbs), 0)
242 242
243 243 def test_rename(self):
244 244 resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
245 245 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
246 246 self.assertEqual(resp.json()['name'], 'z.ipynb')
247 247 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
248 248
249 249 nbs = notebooks_only(self.nb_api.list('foo').json())
250 250 nbnames = set(n['name'] for n in nbs)
251 251 self.assertIn('z.ipynb', nbnames)
252 252 self.assertNotIn('a.ipynb', nbnames)
253 253
254 254 def test_rename_existing(self):
255 255 with assert_http_error(409):
256 256 self.nb_api.rename('a.ipynb', 'foo', 'b.ipynb')
257 257
258 258 def test_save(self):
259 259 resp = self.nb_api.read('a.ipynb', 'foo')
260 260 nbcontent = json.loads(resp.text)['content']
261 261 nb = to_notebook_json(nbcontent)
262 262 ws = new_worksheet()
263 263 nb.worksheets = [ws]
264 264 ws.cells.append(new_heading_cell(u'Created by test Β³'))
265 265
266 266 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
267 267 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
268 268
269 269 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
270 270 with io.open(nbfile, 'r', encoding='utf-8') as f:
271 271 newnb = read(f, format='ipynb')
272 272 self.assertEqual(newnb.worksheets[0].cells[0].source,
273 273 u'Created by test Β³')
274 274 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
275 275 newnb = to_notebook_json(nbcontent)
276 276 self.assertEqual(newnb.worksheets[0].cells[0].source,
277 277 u'Created by test Β³')
278 278
279 279 # Save and rename
280 280 nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
281 281 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
282 282 saved = resp.json()
283 283 self.assertEqual(saved['name'], 'a2.ipynb')
284 284 self.assertEqual(saved['path'], 'foo/bar')
285 285 assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
286 286 assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
287 287 with assert_http_error(404):
288 288 self.nb_api.read('a.ipynb', 'foo')
289 289
290 290 def test_checkpoints(self):
291 291 resp = self.nb_api.read('a.ipynb', 'foo')
292 292 r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
293 293 self.assertEqual(r.status_code, 201)
294 294 cp1 = r.json()
295 295 self.assertEqual(set(cp1), {'id', 'last_modified'})
296 296 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
297 297
298 298 # Modify it
299 299 nbcontent = json.loads(resp.text)['content']
300 300 nb = to_notebook_json(nbcontent)
301 301 ws = new_worksheet()
302 302 nb.worksheets = [ws]
303 303 hcell = new_heading_cell('Created by test')
304 304 ws.cells.append(hcell)
305 305 # Save
306 306 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
307 307 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
308 308
309 309 # List checkpoints
310 310 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
311 311 self.assertEqual(cps, [cp1])
312 312
313 313 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
314 314 nb = to_notebook_json(nbcontent)
315 315 self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
316 316
317 317 # Restore cp1
318 318 r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['id'])
319 319 self.assertEqual(r.status_code, 204)
320 320 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
321 321 nb = to_notebook_json(nbcontent)
322 322 self.assertEqual(nb.worksheets, [])
323 323
324 324 # Delete cp1
325 325 r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['id'])
326 326 self.assertEqual(r.status_code, 204)
327 327 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
328 328 self.assertEqual(cps, [])
329 329
General Comments 0
You need to be logged in to leave comments. Login now