Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def cell_complete(self, cell, cell_index=None, **kwargs):
"""
Finalize metadata for a cell and save notebook.
Optionally called by engines during execution to finalize the
metadata for a cell and save the notebook to the output path.
"""
end_time = self.now()
if self.log_output:
ceel_num = cell_index + 1 if cell_index is not None else ''
logger.info('Ending Cell {:-<43}'.format(ceel_num))
# Ensure our last cell messages are not buffered by python
sys.stdout.flush()
sys.stderr.flush()
cell.metadata.papermill['end_time'] = end_time.isoformat()
if cell.metadata.papermill.get('start_time'):
start_time = dateutil.parser.parse(cell.metadata.papermill['start_time'])
cell.metadata.papermill['duration'] = (end_time - start_time).total_seconds()
if cell.metadata.papermill['status'] != self.FAILED:
cell.metadata.papermill['status'] = self.COMPLETED
self.save()
if self.pbar:
self.pbar.update(1)
Flag for whether or not to hide input.
cwd : str, optional
Working directory to use when executing the notebook
**kwargs
Arbitrary keyword arguments to pass to the notebook engine
Returns
-------
nb : NotebookNode
Executed notebook object
"""
path_parameters = add_builtin_parameters(parameters)
input_path = parameterize_path(input_path, path_parameters)
output_path = parameterize_path(output_path, path_parameters)
logger.info("Input Notebook: %s" % get_pretty_path(input_path))
logger.info("Output Notebook: %s" % get_pretty_path(output_path))
with local_file_io_cwd():
if cwd is not None:
logger.info("Working directory: {}".format(get_pretty_path(cwd)))
nb = load_notebook_node(input_path)
# Parameterize the Notebook.
if parameters:
nb = parameterize_notebook(nb, parameters, report_mode)
nb = prepare_notebook_metadata(nb, input_path, output_path, report_mode)
if not prepare_only:
# Fetch the kernel name if it's not supplied
kernel_name = kernel_name or nb.metadata.kernelspec.name
def cell_start(self, cell, cell_index=None, **kwargs):
"""
Set and save a cell's start state.
Optionally called by engines during execution to initialize the
metadata for a cell and save the notebook to the output path.
"""
if self.log_output:
ceel_num = cell_index + 1 if cell_index is not None else ''
logger.info('Executing Cell {:-<40}'.format(ceel_num))
cell.metadata.papermill['start_time'] = self.now().isoformat()
cell.metadata.papermill["status"] = self.RUNNING
cell.metadata.papermill['exception'] = False
self.save()
Arbitrary keyword arguments to pass to the notebook engine
Returns
-------
nb : NotebookNode
Executed notebook object
"""
path_parameters = add_builtin_parameters(parameters)
input_path = parameterize_path(input_path, path_parameters)
output_path = parameterize_path(output_path, path_parameters)
logger.info("Input Notebook: %s" % get_pretty_path(input_path))
logger.info("Output Notebook: %s" % get_pretty_path(output_path))
with local_file_io_cwd():
if cwd is not None:
logger.info("Working directory: {}".format(get_pretty_path(cwd)))
nb = load_notebook_node(input_path)
# Parameterize the Notebook.
if parameters:
nb = parameterize_notebook(nb, parameters, report_mode)
nb = prepare_notebook_metadata(nb, input_path, output_path, report_mode)
if not prepare_only:
# Fetch the kernel name if it's not supplied
kernel_name = kernel_name or nb.metadata.kernelspec.name
# Execute the Notebook in `cwd` if it is set
with chdir(cwd):
nb = papermill_engines.execute_notebook_with_engine(