thht / plus_sync / 212

Committed 04 May 2026 07:00AM UTC coverage: 64.842% (-27.2%) from 92.008%

Build # 212

Build Type

gitlab-ci

Committed by

Commit Message

update deps

Pull Request Pull Request #32: update deps

Coverage Stats

616 of 950 relevant lines covered (64.84%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.61

/src/plus_sync/cmd/commands/sync.py

from pathlib import Path
from typing import Annotated, Optional

import typer
from tqdm import tqdm

from ...config import Config
from ..app import app
from ..helpers.typer import HashSubjectIDs


@app.command(no_args_is_help=True)
def sync(  # noqa: PLR0913
    remote_name: Annotated[str, typer.Argument(help='The name of the remote to use.')],
    hash_subject_ids: HashSubjectIDs,
    dry_run: Annotated[bool, typer.Option(help='Whether to do a dry run.')] = False,
    limit: Annotated[Optional[int], typer.Option(help='Limit the number of subjects to sync.')] = None,
    only_subject: Annotated[
        Optional[list[str]], typer.Option(help='Only sync the specified subject. Can be used multiple times')
    ] = None,
    sync_folder: Annotated[
        Optional[Path], typer.Option(help='The directory to sync to. If set overrides sync_folder')
    ] = None,
) -> None:
    """
    Sync the data.
    """
    typer.echo('Syncing the data.')
    config = Config.from_cmdargs()
    sync = config.get_sync_by_name(remote_name)
    all_subjects = sync.get_all_subjects(hash=False)
    all_subjects_hashed = sync.get_all_subjects(hash=True)
    if limit:
        all_subjects = list(all_subjects)[:limit]

    if only_subject is not None:
        list_to_test = all_subjects_hashed if hash_subject_ids else all_subjects
        for cur_subject in only_subject:
            if cur_subject not in list_to_test:
                typer.echo(f'Subject {cur_subject} not found in the data.')
                raise typer.Exit(code=1)

        if hash_subject_ids:
            hashed_subjects_idx = [all_subjects_hashed.index(x) for x in only_subject]
            all_subjects = [all_subjects[x] for x in hashed_subjects_idx]
        else:
            all_subjects = only_subject

    all_files_subjects = sync.get_files_for_subjects(all_subjects)
    orig_files_mapping = {
        x: sync.get_local_file(x, hash_subject_ids=hash_subject_ids, sync_folder=sync_folder)
        for x in all_files_subjects
    }
    orig_files_n = len(orig_files_mapping)
    files_to_sync = {remote: local for remote, local in orig_files_mapping.items() if not remote.matches(local)}
    to_sync_n = len(files_to_sync)

    typer.echo(f'Found {orig_files_n} files, {to_sync_n} need syncing.')

    for remote_file, local_file in tqdm(files_to_sync.items()):
        local_path = Path(local_file.path)
        if dry_run:
            typer.echo(f'Would sync {remote_file.path} to {local_path}.')
        else:
            content = sync.get_content(remote_file)
            if not local_path.parent.exists():
                local_path.parent.mkdir(parents=True)
            with Path(local_path).open('wb') as f:
                f.write(content)

1	from pathlib import Path	×
2	from typing import Annotated, Optional	×
3
4	import typer	×
5	from tqdm import tqdm	×
6
7	from ...config import Config	×
8	from ..app import app	×
9	from ..helpers.typer import HashSubjectIDs	×
10
11
12	@app.command(no_args_is_help=True)	×
13	def sync( # noqa: PLR0913	×
14	remote_name: Annotated[str, typer.Argument(help='The name of the remote to use.')],
15	hash_subject_ids: HashSubjectIDs,
16	dry_run: Annotated[bool, typer.Option(help='Whether to do a dry run.')] = False,
17	limit: Annotated[Optional[int], typer.Option(help='Limit the number of subjects to sync.')] = None,
18	only_subject: Annotated[
19	Optional[list[str]], typer.Option(help='Only sync the specified subject. Can be used multiple times')
20	] = None,
21	sync_folder: Annotated[
22	Optional[Path], typer.Option(help='The directory to sync to. If set overrides sync_folder')
23	] = None,
24	) -> None:
25	"""
26	Sync the data.
27	"""
28	typer.echo('Syncing the data.')	1✔
29	config = Config.from_cmdargs()	1✔
30	sync = config.get_sync_by_name(remote_name)	1✔
31	all_subjects = sync.get_all_subjects(hash=False)	1✔
32	all_subjects_hashed = sync.get_all_subjects(hash=True)	1✔
33	if limit:	1✔
34	all_subjects = list(all_subjects)[:limit]	1✔
35
36	if only_subject is not None:	1✔
37	list_to_test = all_subjects_hashed if hash_subject_ids else all_subjects	1✔
38	for cur_subject in only_subject:	1✔
39	if cur_subject not in list_to_test:	1✔
40	typer.echo(f'Subject {cur_subject} not found in the data.')	1✔
41	raise typer.Exit(code=1)	1✔
42
43	if hash_subject_ids:	1✔
44	hashed_subjects_idx = [all_subjects_hashed.index(x) for x in only_subject]	1✔
45	all_subjects = [all_subjects[x] for x in hashed_subjects_idx]	1✔
46	else:
47	all_subjects = only_subject	1✔
48
49	all_files_subjects = sync.get_files_for_subjects(all_subjects)	1✔
50	orig_files_mapping = {	1✔
51	x: sync.get_local_file(x, hash_subject_ids=hash_subject_ids, sync_folder=sync_folder)
52	for x in all_files_subjects
53	}
54	orig_files_n = len(orig_files_mapping)	1✔
55	files_to_sync = {remote: local for remote, local in orig_files_mapping.items() if not remote.matches(local)}	1✔
56	to_sync_n = len(files_to_sync)	1✔
57
58	typer.echo(f'Found {orig_files_n} files, {to_sync_n} need syncing.')	1✔
59
60	for remote_file, local_file in tqdm(files_to_sync.items()):	1✔
61	local_path = Path(local_file.path)	1✔
62	if dry_run:	1✔
63	typer.echo(f'Would sync {remote_file.path} to {local_path}.')	×
64	else:
65	content = sync.get_content(remote_file)	1✔
66	if not local_path.parent.exists():	1✔
67	local_path.parent.mkdir(parents=True)	1✔
68	with Path(local_path).open('wb') as f:	1✔
69	f.write(content)	1✔

thht / plus_sync / 212

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous