Coverage for databooks/cli.py: 89%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Main CLI application."""
2from importlib.metadata import metadata
3from pathlib import Path
4from typing import List, Optional
6from rich.progress import (
7 BarColumn,
8 Progress,
9 SpinnerColumn,
10 TextColumn,
11 TimeElapsedColumn,
12)
13from typer import Argument, BadParameter, Exit, Option, Typer, echo
15from databooks.common import expand_paths, get_logger
16from databooks.conflicts import conflicts2nbs, path2conflicts
17from databooks.metadata import clear_all
19_DISTRIBUTION_METADATA = metadata("databooks")
21logger = get_logger(__file__)
23app = Typer()
26def version_callback(value: bool) -> None:
27 """Return application version."""
28 if value:
29 echo("databooks version: " + _DISTRIBUTION_METADATA["Version"])
30 raise Exit()
33@app.callback()
34def callback( # noqa: D103
35 version: Optional[bool] = Option(
36 None, "--version", callback=version_callback, is_eager=True
37 )
38) -> None:
39 ...
42# add docs dynamically from `pyproject.toml`
43callback.__doc__ = _DISTRIBUTION_METADATA["Summary"]
46@app.command()
47def meta(
48 paths: List[Path] = Argument(..., help="Path(s) of notebook files"),
49 ignore: List[str] = Option(["!*"], help="Glob expression(s) of files to ignore"),
50 prefix: str = Option("", help="Prefix to add to filepath when writing files"),
51 suffix: str = Option("", help="Suffix to add to filepath when writing files"),
52 rm_outs: bool = Option(False, help="Whether to remove cell outputs"),
53 rm_exec: bool = Option(True, help="Whether to remove the cell execution counts"),
54 nb_meta_keep: List[str] = Option([], help="Notebook metadata fields to keep"),
55 cell_meta_keep: List[str] = Option([], help="Cells metadata fields to keep"),
56 overwrite: bool = Option(
57 False, "--overwrite", "-w", help="Confirm overwrite of files"
58 ),
59 check: bool = Option(
60 False,
61 "--check",
62 help="Don't write files but check whether there is unwanted metadata",
63 ),
64 verbose: bool = Option(
65 False, "--verbose", "-v", help="Log processed files in console"
66 ),
67) -> None:
68 """Clear both notebook and cell metadata."""
69 if any(path.suffix not in ("", ".ipynb") for path in paths):
70 raise BadParameter(
71 "Expected either notebook files, a directory or glob expression."
72 )
73 nb_paths = expand_paths(paths=paths, ignore=ignore)
74 if not bool(prefix + suffix) and not check:
75 if not overwrite:
76 raise BadParameter(
77 "No prefix nor suffix were passed."
78 " Please specify `--overwrite` or `-w` to overwrite files."
79 )
80 else:
81 logger.warning(f"{len(nb_paths)} files will be overwritten")
83 write_paths = [p.parent / (prefix + p.stem + suffix + p.suffix) for p in nb_paths]
84 with Progress(
85 SpinnerColumn(),
86 TextColumn("[progress.description]{task.description}"),
87 BarColumn(),
88 TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
89 TimeElapsedColumn(),
90 ) as progress:
91 metadata = progress.add_task("[yellow]Removing metadata", total=len(nb_paths))
93 are_equal = clear_all(
94 read_paths=nb_paths,
95 write_paths=write_paths,
96 progress_callback=lambda: progress.update(metadata, advance=1),
97 notebook_metadata_keep=nb_meta_keep,
98 cell_metadata_keep=cell_meta_keep,
99 cell_execution_count=rm_exec,
100 cell_outputs=rm_outs,
101 check=check,
102 verbose=verbose,
103 )
104 if check:
105 if all(are_equal):
106 logger.info("No unwanted metadata!")
107 else:
108 logger.info(
109 f"Found unwanted metadata in {sum(not eq for eq in are_equal)} out of"
110 f" {len(are_equal)} files"
111 )
112 raise Exit(code=1)
113 else:
114 logger.info(
115 f"The metadata of {sum(not eq for eq in are_equal)} out of {len(are_equal)}"
116 " notebooks were removed!"
117 )
120@app.command()
121def fix(
122 paths: List[Path] = Argument(..., help="Path(s) of notebook files with conflicts"),
123 ignore: List[str] = Option(["!*"], help="Glob expression(s) of files to ignore"),
124 metadata_first: bool = Option(
125 True, help="Whether or not to keep the metadata from the first/current notebook"
126 ),
127 cells_first: Optional[bool] = Option(
128 None,
129 help="Whether to keep the cells from the first or last notebook."
130 " Omit to keep both",
131 ),
132 interactive: bool = Option(
133 False,
134 "--interactive",
135 "-i",
136 help="Interactively resolve the conflicts (not implemented)",
137 ),
138 verbose: bool = Option(False, help="Log processed files in console"),
139) -> None:
140 """
141 Fix git conflicts for notebooks.
143 Perform by getting the unmerged blobs from git index, comparing them and returning
144 a valid notebook summarizing the differences - see
145 [git docs](https://git-scm.com/docs/git-ls-files).
146 """
147 filepaths = expand_paths(paths=paths, ignore=ignore)
148 conflict_files = path2conflicts(nb_paths=filepaths)
149 if not conflict_files:
150 raise BadParameter(
151 f"No conflicts found at {', '.join([str(p) for p in filepaths])}."
152 )
153 if interactive:
154 raise NotImplementedError
156 with Progress(
157 SpinnerColumn(),
158 TextColumn("[progress.description]{task.description}"),
159 BarColumn(),
160 TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
161 TimeElapsedColumn(),
162 ) as progress:
163 conflicts = progress.add_task(
164 "[yellow]Removing metadata", total=len(conflict_files)
165 )
166 conflicts2nbs(
167 conflict_files=conflict_files,
168 keep_first=metadata_first,
169 cells_first=cells_first,
170 verbose=verbose,
171 progress_callback=lambda: progress.update(conflicts, advance=1),
172 )
173 logger.info(f"Resolved the conflicts of {len(conflict_files)}!")
176@app.command()
177def diff() -> None:
178 """Show differences between notebooks (not implemented)."""
179 raise NotImplementedError