Coverage for b4_backup/cli/main.py: 100%
94 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-18 22:40 +0000
1"""Contains code for the main part of the CLI."""
3import logging
5import typer
6from rich import prompt
8from b4_backup import exceptions
9from b4_backup.cli.init import app
10from b4_backup.cli.tools import app as tools_app
11from b4_backup.cli.utils import (
12 OutputFormat,
13 complete_target,
14 error_handler,
15 validate_target,
16)
17from b4_backup.config_schema import BaseConfig, TargetRestoreStrategy
18from b4_backup.main.b4_backup import B4Backup
19from b4_backup.main.backup_target_host import host_generator
20from b4_backup.main.dataclass import ChoiceSelector
22log = logging.getLogger("b4_backup.cli")
24app.add_typer(tools_app, name="tools")
27@app.command()
28def backup(
29 ctx: typer.Context,
30 target: list[str] = typer.Option(
31 [],
32 "-t",
33 "--target",
34 help="Selected targets to backup",
35 autocompletion=complete_target,
36 callback=validate_target,
37 ),
38 name: str = typer.Option(
39 "manual",
40 "-n",
41 "--name",
42 help="Name suffix (and retention ruleset) for this backup",
43 ),
44 source_only: bool = typer.Option(
45 False,
46 help="Perform actions on source side only",
47 ),
48):
49 """Perform backups on specified targets. If no target is specified, the default targets defined in the config will be used."""
50 config: BaseConfig = ctx.obj
51 target_choice = ChoiceSelector(target or config.default_targets)
53 b4_backup = B4Backup(config.timezone)
55 with error_handler() as err_handler:
56 snapshot_name = b4_backup.generate_snapshot_name(name)
58 for src_host, dst_host in host_generator(
59 target_choice, config.backup_targets, use_destination=not source_only
60 ):
61 try:
62 if not src_host:
63 raise exceptions.InvalidConnectionUrlError( # noqa: TRY301
64 "Backup requires source to be specified"
65 )
67 b4_backup.backup(src_host, dst_host, snapshot_name)
68 except Exception as exc:
69 err_handler.add(exc)
72@app.command(name="list")
73def list_snapshots(
74 ctx: typer.Context,
75 target: list[str] = typer.Option(
76 [],
77 "-t",
78 "--target",
79 help="Selected targets to backup",
80 autocompletion=complete_target,
81 callback=validate_target,
82 ),
83 source: bool = typer.Option(False, help="List snapshots on source host"),
84 destination: bool = typer.Option(False, help="List snapshots on destination host"),
85 format: OutputFormat = typer.Option(OutputFormat.RICH.value, help="Output format"),
86):
87 """List all snapshots for the specified targets."""
88 config: BaseConfig = ctx.obj
89 target_choice = ChoiceSelector(target or config.default_targets)
90 with error_handler():
91 for src_host, dst_host in host_generator(
92 target_choice,
93 config.backup_targets,
94 use_source=source,
95 use_destination=destination,
96 ):
97 if src_host:
98 OutputFormat.output(src_host.snapshots(), "Source", format)
99 if dst_host:
100 OutputFormat.output(dst_host.snapshots(), "Destination", format)
103@app.command()
104def clean(
105 ctx: typer.Context,
106 target: list[str] = typer.Option(
107 [],
108 "-t",
109 "--target",
110 help="Selected targets to backup",
111 autocompletion=complete_target,
112 callback=validate_target,
113 ),
114 source_only: bool = typer.Option(
115 False,
116 help="Perform actions on source side only",
117 ),
118):
119 """Apply the targets retention ruleset without performing a backup."""
120 config: BaseConfig = ctx.obj
121 target_choice = ChoiceSelector(target or config.default_targets)
123 b4_backup = B4Backup(config.timezone)
125 with error_handler():
126 for src_host, dst_host in host_generator(
127 target_choice, config.backup_targets, use_destination=not source_only
128 ):
129 if not src_host:
130 raise exceptions.InvalidConnectionUrlError("Clean requires source to be specified")
132 b4_backup.clean(src_host, dst_host)
135@app.command()
136def delete(
137 ctx: typer.Context,
138 snapshot_name: str = typer.Argument(..., help="Name of the snapshot you want to restore"),
139 target: list[str] = typer.Option(
140 [],
141 "-t",
142 "--target",
143 help="Selected targets to backup",
144 autocompletion=complete_target,
145 callback=validate_target,
146 ),
147 source: bool = typer.Option(False, help="Delete from source host"),
148 destination: bool = typer.Option(False, help="Delete from destination host"),
149):
150 """Delete a specific snapshot from the source and/or destination."""
151 config: BaseConfig = ctx.obj
152 target_choice = ChoiceSelector(target or config.default_targets)
153 b4_backup = B4Backup(config.timezone)
154 with error_handler():
155 for src_host, dst_host in host_generator(
156 target_choice, config.backup_targets, use_source=source, use_destination=destination
157 ):
158 if src_host:
159 b4_backup.delete(src_host, snapshot_name)
160 if dst_host:
161 b4_backup.delete(dst_host, snapshot_name)
164@app.command()
165def delete_all(
166 ctx: typer.Context,
167 target: list[str] = typer.Option(
168 [],
169 "-t",
170 "--target",
171 help="Selected targets to backup",
172 autocompletion=complete_target,
173 callback=validate_target,
174 ),
175 retention: list[str] = typer.Option(
176 ["ALL"],
177 "-r",
178 "--retention",
179 help="Name suffix (and retention ruleset) for this backup",
180 ),
181 force: bool = typer.Option(False, help="Skip confirmation prompt"),
182 source: bool = typer.Option(False, help="Delete from source host"),
183 destination: bool = typer.Option(False, help="Delete from destination host"),
184):
185 """Delete all local and remote backups of the specified target/retention ruleset combination. Equivalent to an "all: 0" rule."""
186 config: BaseConfig = ctx.obj
187 target_choice = ChoiceSelector(target or config.default_targets)
188 retention_names = ChoiceSelector(retention)
190 log.warning(
191 "You are about to DELETE all snapshots with these retention_names (%s) for these targets: %s",
192 ", ".join(retention),
193 ", ".join(target_choice.resolve_target(config.backup_targets)),
194 )
195 if not force and not prompt.Confirm.ask("Continue"):
196 raise typer.Exit(1)
198 b4_backup = B4Backup(config.timezone)
200 with error_handler():
201 for src_host, dst_host in host_generator(
202 target_choice, config.backup_targets, use_source=source, use_destination=destination
203 ):
204 if src_host:
205 b4_backup.delete_all(src_host, retention_names)
207 if dst_host:
208 b4_backup.delete_all(dst_host, retention_names)
211@app.command()
212def restore(
213 ctx: typer.Context,
214 snapshot_name: str = typer.Argument(..., help="Name of the snapshot you want to restore"),
215 target: list[str] = typer.Option(
216 [],
217 "-t",
218 "--target",
219 help="Selected targets to backup",
220 autocompletion=complete_target,
221 callback=validate_target,
222 ),
223 strategy: TargetRestoreStrategy | None = typer.Option(
224 None,
225 help="Restore strategy or procedure to apply",
226 ),
227 source_only: bool = typer.Option(
228 False,
229 help="Perform actions on source side only",
230 ),
231):
232 """
233 Restore one or more targets based on a previously created snapshot.
234 You can revert a REPLACE restore by using REPLACE als snapshot name and strategy.
235 """
236 config: BaseConfig = ctx.obj
237 target_choice = ChoiceSelector(target or config.default_targets)
239 b4_backup = B4Backup(config.timezone)
241 with error_handler():
242 for src_host, dst_host in host_generator(
243 target_choice, config.backup_targets, use_destination=not source_only
244 ):
245 if not src_host:
246 raise exceptions.InvalidConnectionUrlError(
247 "Restore requires source to be specified"
248 )
250 target_strategy = strategy or src_host.target_config.restore_strategy
252 b4_backup.restore(src_host, dst_host, snapshot_name, target_strategy)
255@app.command()
256def sync(
257 ctx: typer.Context,
258 target: list[str] = typer.Option(
259 [],
260 "-t",
261 "--target",
262 help="Selected targets to backup",
263 autocompletion=complete_target,
264 callback=validate_target,
265 ),
266):
267 """Send pending snapshots to the destination."""
268 config: BaseConfig = ctx.obj
269 target_choice = ChoiceSelector(target or config.default_targets)
271 b4_backup = B4Backup(config.timezone)
273 with error_handler():
274 for src_host, dst_host in host_generator(target_choice, config.backup_targets):
275 if not src_host or not dst_host:
276 raise exceptions.InvalidConnectionUrlError(
277 "Sync requires source and destination to be specified"
278 )
280 b4_backup.sync(src_host, dst_host)
283# A collection of stuff I would like to improve
285## Tooling
286# TODO: Replace poetry with uv
288## Features
289# TODO: pre/post backup hooks
290# - Option to execute code before and after an update
291# Would be handy for enabling maintanance mode during an update
293## Documentation
294# TODO: Docs
295# - Auto create links to reference and terminology
297## Visual CLI and logging improvements
298# TODO: b4 cmd without any options should show --help text
299# TODO: rich print_log function
300# - To optionally print logs like a standard rich.print()
301# - Will be in a seperate logger. Maybe b4_backup.print
302# - Subvolume transmission info (Bytes transfered, duration, rate)
304# TODO: List snapshots flags
305# - Show status of snapshot for example: stale, async
307# TODO: b4 switch command to switch to version:
308# - b4 switch source to go to the source directory if available
309# - Needs to be executed using "cd $(b4 switch 2024-05-21-14-41-10_manual)"
310# - A better approach would be a modification in bashrc/zshrc
311# - Implement autocomplete for snapshots