Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def execute_managed_notebook(
cls,
nb_man,
kernel_name,
log_output=False,
start_timeout=60,
execution_timeout=None,
**kwargs
):
# Nicely handle preprocessor arguments prioritizing values set by engine
preprocessor = DagstermillExecutePreprocessor(
timeout=execution_timeout if execution_timeout else kwargs.get('timeout'),
startup_timeout=start_timeout,
kernel_name=kernel_name,
log=logger,
)
preprocessor.log_output = log_output # pylint:disable = attribute-defined-outside-init
preprocessor.preprocess(nb_man, kwargs)
Note: The preprocessor concept in this method is similar to what is used
by `nbconvert`, and it is somewhat misleading here. The preprocesser
represents a notebook processor, not a preparation object.
"""
# Exclude parameters that named differently downstream
safe_kwargs = remove_args(['timeout', 'startup_timeout'], **kwargs)
# Nicely handle preprocessor arguments prioritizing values set by engine
final_kwargs = merge_kwargs(
safe_kwargs,
timeout=execution_timeout if execution_timeout else kwargs.get('timeout'),
startup_timeout=start_timeout,
kernel_name=kernel_name,
log=logger,
log_output=log_output,
stdout_file=stdout_file,
stderr_file=stderr_file,
)
preprocessor = PapermillExecutePreprocessor(**final_kwargs)
preprocessor.preprocess(nb_man, safe_kwargs)
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