Newer
Older
# This has to be using ._git.commit to raise an error if no results have been written.
commit_return = self.output_repo._git.commit("-m", message)
self.update_output_master_logs()
print("\n" + commit_return + "\n")
except git.exc.GitCommandError as e:
self.output_repo.delete_active_branch_if_branch_is_empty()
raise e
finally:
self._is_in_context_manager = False
self._on_context_enter_commit_hash = None
@contextlib.contextmanager
def track_results(self, results_commit_message: str, debug=False):
Context manager to be used when running project code that produces output that should
be tracked in the output repository.
:param results_commit_message:
Commit message for the commit of the output repository.
"""
if debug:
yield "debug"
return
new_branch_name = self.enter_context()
try:
yield new_branch_name
except Exception as e:
self.output_repo.delete_active_branch_if_branch_is_empty()
raise e
else:
self.exit_context(message=results_commit_message)
class OutputRepo(BaseRepo):
pass
class JupyterInterfaceRepo(ProjectRepo):
def commit(self, message: str, add_all=True):
Notebook.save_ipynb()
super().commit(message, add_all)
def commit_nb_output(self, notebook_path: str, results_commit_message: str,
force_rerun=False, timeout=600, conversion_formats: list = None):
if not Path(notebook_path).is_absolute():
notebook_path = self.working_dir / notebook_path
notebook = Notebook(notebook_path)
if "nbconvert_call" not in sys.argv:
# This is reached in the first call of this function
self.enter_context(force=False)
notebook.check_and_rerun_notebook(force_rerun=force_rerun,
timeout=timeout)
else:
# This is executed during the nbconvert call
notebook.convert_ipynb(self.output_path, formats=conversion_formats)
notebook.export_all_figures(self.output_path)
self.exit_context(results_commit_message)