28399320195

Committed 29 Jun 2026 08:05PM UTC coverage: 55.337%. First build

Build # 28399320195

Build Type

Pull #159

github

Committed by

web-flow

Commit Message

Merge ff620e746 into 1c02a8c92

Pull Request Pull Request #159: Obsidian Tasks syncronize with Google Tasks

Coverage Stats

87 of 266 new or added lines in 5 files covered. (32.71%)

1804 of 3260 relevant lines covered (55.34%)

0.55 hits per line

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

81.99

/syncall/cli.py

"""CLI argument functions - reuse across your apps.

This module will be loaded regardless of extras - don't put something here that requires an
extra dependency.
"""

import os
import sys

import click
from bubop import format_list, logger

from syncall import version
from syncall.app_utils import (
    determine_app_config_fname,
    error_and_exit,
    fetch_from_pass_manager,
    get_named_combinations,
    name_to_resolution_strategy_type,
)
from syncall.constants import COMBINATION_FLAGS
from syncall.pdb_cli_utils import run_pdb_on_error as _run_pdb_on_error


def _set_own_excepthook(ctx, param, value):
    del param

    if not value or ctx.resilient_parsing:
        return value

    sys.excepthook = _run_pdb_on_error
    return value


def _opt_pdb_on_error():
    return click.option(
        "--pdb-on-error",
        "pdb_on_error",
        is_flag=True,
        help="Invoke PDB if there's an uncaught exception during the program execution.",
        callback=_set_own_excepthook,
        expose_value=True,
        is_eager=True,
    )


# asana related options -----------------------------------------------------------------------
def opts_asana(hidden_gid: bool):
    def decorator(f):
        for d in reversed(
            [
                _opt_asana_token_pass_path,
                _opt_asana_workspace_gid,
                _opt_asana_workspace_name,
                _opt_list_asana_workspaces,
            ],
        ):
            f = d()(f)

        # --asana-task-gid is used to ease development and debugging. It is not currently
        # suitable for regular use.
        return _opt_asana_task_gid(hidden=hidden_gid)(f)

    return decorator


def _opt_asana_task_gid(**kwargs):
    return click.option(
        "-a",
        "--asana-task-gid",
        "asana_task_gid",
        type=str,
        help="Limit sync to provided task",
        **kwargs,
    )


def _opt_asana_token_pass_path():
    def callback(ctx, param, value):
        del ctx

        api_token_pass_path = value

        # fetch API token to connect to asana -------------------------------------------------
        asana_token = os.environ.get("ASANA_PERSONAL_ACCESS_TOKEN")

        if asana_token is None and api_token_pass_path is None:
            error_and_exit(
                "You must provide an Asana Personal Access asana_token, using the"
                f" {'/'.join(param.opts)} option",
            )
        if asana_token is not None:
            logger.debug(
                "Reading the Asana Personal Access asana_token (PAT) from environment"
                " variable...",
            )
        else:
            asana_token = fetch_from_pass_manager(api_token_pass_path)

        return asana_token

    return click.option(
        "--token",
        "--token-pass-path",
        "asana_token",
        help="Path in the UNIX password manager to fetch",
        expose_value=True,
        callback=callback,
    )


def _opt_asana_workspace_gid():
    return click.option(
        "-w",
        "--asana-workspace-gid",
        "asana_workspace_gid",
        type=str,
        help="Asana workspace GID used to filter tasks",
    )


def _opt_asana_workspace_name():
    return click.option(
        "-W",
        "--asana-workspace-name",
        "asana_workspace_name",
        type=str,
        help="Asana workspace name used to filter tasks",
    )


def _opt_list_asana_workspaces():
    return click.option(
        "--list-asana-workspaces",
        "do_list_asana_workspaces",
        is_flag=True,
        help="List the available Asana workspaces",
    )


# taskwarrior options -------------------------------------------------------------------------
def opts_tw_filtering():
    def decorator(f):
        for d in reversed(
            [
                _opt_tw_filter,
                _opt_tw_all_tasks,
                _opt_tw_tags,
                _opt_tw_project,
                _opt_tw_only_tasks_modified_X_days,
                _opt_prefer_scheduled_date,
            ],
        ):
            f = d()(f)
        return f

    return decorator


def _opt_tw_filter():
    return click.option(
        "-f",
        "--tw-filter",
        "tw_filter",
        type=str,
        help=(
            "Taskwarrior filter for specifying the tasks to synchronize. These filters will be"
            " concatenated using AND with potential tags and projects potentially specified."
        ),
    )


def _opt_tw_all_tasks():
    return click.option(
        "--all",
        "--taskwarrior-all-tasks",
        "tw_sync_all_tasks",
        is_flag=True,
        help="Sync all taskwarrior tasks (potentially very slow).",
    )


def _opt_tw_tags():
    return click.option(
        "-t",
        "--taskwarrior-tags",
        "tw_tags",
        type=str,
        help="Taskwarrior tags to synchronize.",
        expose_value=True,
        multiple=True,
    )


def _opt_tw_project():
    return click.option(
        "-p",
        "--tw-project",
        "tw_project",
        type=str,
        help="Taskwarrior project to synchronize.",
        expose_value=True,
        is_eager=True,
    )


def _opt_tw_only_tasks_modified_X_days():
    def callback(ctx, param, value):
        del param

        if value is None or ctx.resilient_parsing:
            return None

        return f"modified.after:-{value}d"

    return click.option(
        "--days",
        "--only-modified-last-X-days",
        "tw_only_modified_last_X_days",
        type=str,
        help=(
            "Only synchronize Taskwarrior tasks that have been modified in the last X days"
            " (specify X, e.g., 1, 30, 0.5, etc.)."
        ),
        callback=callback,
    )


def _opt_prefer_scheduled_date():
    return click.option(
        "--prefer-scheduled-date",
        "prefer_scheduled_date",
        is_flag=True,
        help=(
            'Prefer using the "scheduled" date field instead of the "due" date if the former'
            " is available."
        ),
    )


# notion --------------------------------------------------------------------------------------
def opt_notion_page_id():
    return click.option(
        "-n",
        "--notion-page",
        "notion_page_id",
        type=str,
        help="Page ID of the Notion page to synchronize.",
    )


def opt_notion_token_pass_path():
    return click.option(
        "--token",
        "--token-pass-path",
        "token_pass_path",
        help="Path in the UNIX password manager to fetch.",
    )


# gkeep ---------------------------------------------------------------------------------------
def opts_gkeep():
    def decorator(f):
        for d in reversed(
            [
                _opt_gkeep_user_pass_path,
                _opt_gkeep_passwd_pass_path,
                _opt_gkeep_token_pass_path,
            ],
        ):
            f = d()(f)

        return f

    return decorator


def _opt_gkeep_user_pass_path():
    return click.option(
        "--user",
        "--user-pass-path",
        "gkeep_user_pass_path",
        help="Path in the UNIX password manager to fetch the Google username from.",
        default="gkeepapi/user",
    )


def _opt_gkeep_passwd_pass_path():
    return click.option(
        "--passwd",
        "--passwd-pass-path",
        "gkeep_passwd_pass_path",
        help="Path in the UNIX password manager to fetch the Google password from.",
        default="gkeepapi/passwd",
    )


def _opt_gkeep_token_pass_path():
    return click.option(
        "--token",
        "--token-pass-path",
        "gkeep_token_pass_path",
        help="Path in the UNIX password manager to fetch the google keep token from.",
        default="gkeepapi/token",
    )


def opt_gkeep_labels():
    return click.option(
        "-k",
        "--gkeep-labels",
        type=str,
        multiple=True,
        help="Google Keep labels whose notes to synchronize.",
    )


def opt_gkeep_ignore_labels():
    return click.option(
        "-i",
        "--gkeep-ignore-labels",
        type=str,
        multiple=True,
        help="Google Keep labels whose notes will be explicitly ignored.",
    )


def opt_gkeep_note():
    return click.option(
        "-k",
        "--gkeep-note",
        type=str,
        help=(
            "Full title of the Google Keep Note to synchronize - Make sure you enable the"
            " checkboxes."
        ),
    )


# google calendar -----------------------------------------------------------------------------
def opt_gcal_calendar():
    return click.option(
        "-c",
        "--gcal-calendar",
        type=str,
        help="Name of the Google Calendar to synchronize (will be created if not there).",
    )


# google tasks --------------------------------------------------------------------------------
def opt_gtasks_list():
    return click.option(
        "-l",
        "--gtasks-list",
        type=str,
        help="Name of the Google Tasks list to synchronize (will be created if not there).",
    )


# markdown file -------------------------------------------------------------------------------
def opts_markdown():
    def decorator(f):
        for d in reversed(
            [
                _opt_md_file,
                _opt_prefer_scheduled_date,
            ],
        ):
            f = d()(f)
        return f

    return decorator


def _opt_md_file():
    return click.option(
        "-m",
        "--markdown-file",
        type=str,
        help="Name of the Markdown file including tasks list to synchronize",
    )


# google-related options ----------------------------------------------------------------------
def opt_google_secret_override():
    return click.option(
        "--google-secret",
        default=None,
        type=click.Path(exists=True, file_okay=True, dir_okay=False, resolve_path=True),
        help="Override the client secret used for the communication with the Google APIs.",
    )


def opt_google_oauth_port():
    return click.option(
        "--oauth-port",
        default=8081,
        type=int,
        help="Port to use for OAuth Authentication with Google Applications.",
    )


# caldav options ------------------------------------------------------------------------------
def opts_caldav():
    def decorator(f):
        for d in reversed(
            [
                _opt_caldav_calendar,
                _opt_caldav_url,
                _opt_caldav_user,
                _opt_caldav_passwd_pass_path,
                _opt_caldav_passwd_cmd,
            ],
        ):
            f = d()(f)

        return f

    return decorator


def _opt_caldav_calendar():
    return click.option(
        "--caldav-calendar",
        type=str,
        default="Personal",
        help="Name of the caldav Calendar to sync (will be created if not there).",
    )


def _opt_caldav_url():
    return click.option(
        "--caldav-url",
        type=str,
        help="URL where the caldav calendar is hosted at (including /dav if applicable).",
    )


def _opt_caldav_user():
    return click.option(
        "--caldav-user",
        "caldav_user",
        help="The caldav username for the given caldav instance",
    )


def _opt_caldav_passwd_pass_path():
    return click.option(
        "--caldav-passwd",
        "--caldav-passwd-pass-path",
        "caldav_passwd_pass_path",
        help="Path in the UNIX password manager to fetch the caldav password from.",
    )


def _opt_caldav_passwd_cmd():
    return click.option(
        "--caldav-passwd-cmd",
        "caldav_passwd_cmd",
        help="Command that outputs the caldav password on stdout.",
    )


# filesystem related options ------------------------------------------------------------------
def opt_filesystem_root():
    return click.option(
        "--fs",
        "--fs-root",
        "filesystem_root",
        required=False,
        type=str,
        help="Directory to consider as root for synchronization operations.",
    )


def opt_filename_extension():
    return click.option(
        "--ext",
        "--filename-extension",
        "filename_extension",
        type=str,
        help="Use this extension for locally created files.",
        default=".md",
    )

def opt_filename_path():
    return click.option(
        "--path",
        "--filename-path",
        "filename_path",
        type=str,
        help="Use this file path for locally saved data",
    )

# general options -----------------------------------------------------------------------------
def opts_miscellaneous(side_A_name: str, side_B_name: str):
    def decorator(f):
        for d in reversed(
            [
                (_opt_list_resolution_strategies,),
                (_opt_resolution_strategy,),
                (_opt_confirm,),
                (
                    click.version_option,
                    version,
                ),
                (_opt_pdb_on_error,),
                (_opt_list_combinations, side_A_name, side_B_name),
                (_opt_combination, side_A_name, side_B_name),
                (_opt_custom_combination_savename, side_A_name, side_B_name),
            ],
        ):
            fn = d[0]
            fn_args = d[1:]
            f = fn(*fn_args)(f)  # type: ignore

        return click.option("-v", "--verbose", count=True)(f)

    return decorator


def _opt_confirm():
    return click.option(
        "--confirm",
        "confirm",
        is_flag=True,
        default=False,
        expose_value=True,
        help="Confirm app configuration before proceeding with the actual execution",
    )


def opt_default_duration_event_mins():
    return click.option(
        "--default-event-duration-mins",
        "default_event_duration_mins",
        default=30,
        type=int,
        help="The default duration of an event that is to be created [in minutes].",
    )


def _list_named_combinations(config_fname: str) -> None:
    """List the named configurations currently available for the given configuration name."""
    logger.success(
        format_list(
            header="\n\nNamed configurations currently available",
            items=get_named_combinations(config_fname=config_fname),
        ),
    )


def _opt_list_combinations(side_A_name: str, side_B_name: str):
    def callback(ctx, param, value):
        del ctx, param
        if value is True:
            _list_named_combinations(
                config_fname=determine_app_config_fname(
                    side_A_name=side_A_name,
                    side_B_name=side_B_name,
                ),
            )
            sys.exit(0)

    return click.option(
        "--list-combinations",
        "do_list_combinations",
        is_flag=True,
        expose_value=False,
        help=f"List the available named {side_A_name}<->{side_B_name} combinations.",
        callback=callback,
    )


def _opt_resolution_strategy():
    return click.option(
        "-r",
        "--resolution-strategy",
        default="AlwaysSecondRS",
        type=click.Choice(list(name_to_resolution_strategy_type.keys())),
        help="Resolution strategy to use during conflicts.",
    )


def _opt_list_resolution_strategies():
    def _list_resolution_strategies(ctx, param, value):
        del ctx, param

        if value is not True:
            return

        strs = name_to_resolution_strategy_type.keys()
        click.echo(
            "\n".join(
                [f"{a}. {b}" for a, b in zip(range(1, len(strs) + 1), strs, strict=False)],
            ),
        )
        sys.exit(0)

    return click.option(
        "--list-resolution-strategies",
        callback=_list_resolution_strategies,
        is_flag=True,
        expose_value=False,
        help="List all the available resolution strategies and exit.",
    )


def _opt_combination(side_A_name: str, side_B_name: str):
    return click.option(
        COMBINATION_FLAGS[0],
        COMBINATION_FLAGS[1],
        "combination_name",
        type=str,
        help=f"Name of an already saved {side_A_name}<->{side_B_name} combination.",
    )


def _opt_custom_combination_savename(side_A_name: str, side_B_name: str):
    return click.option(
        "-s",
        "--save-as",
        "custom_combination_savename",
        type=str,
        help=(
            f"Save the given {side_A_name}<->{side_B_name} filters combination using a"
            " specified custom name."
        ),
    )

1	"""CLI argument functions - reuse across your apps.
2
3	This module will be loaded regardless of extras - don't put something here that requires an
4	extra dependency.
5	"""
6
7	import os	1✔
8	import sys	1✔
9
10	import click	1✔
11	from bubop import format_list, logger	1✔
12
13	from syncall import version	1✔
14	from syncall.app_utils import (	1✔
15	determine_app_config_fname,
16	error_and_exit,
17	fetch_from_pass_manager,
18	get_named_combinations,
19	name_to_resolution_strategy_type,
20	)
21	from syncall.constants import COMBINATION_FLAGS	1✔
22	from syncall.pdb_cli_utils import run_pdb_on_error as _run_pdb_on_error	1✔
23
24
25	def _set_own_excepthook(ctx, param, value):	1✔
26	del param	×
27
28	if not value or ctx.resilient_parsing:	×
29	return value	×
30
31	sys.excepthook = _run_pdb_on_error	×
32	return value	×
33
34
35	def _opt_pdb_on_error():	1✔
36	return click.option(	1✔
37	"--pdb-on-error",
38	"pdb_on_error",
39	is_flag=True,
40	help="Invoke PDB if there's an uncaught exception during the program execution.",
41	callback=_set_own_excepthook,
42	expose_value=True,
43	is_eager=True,
44	)
45
46
47	# asana related options -----------------------------------------------------------------------
48	def opts_asana(hidden_gid: bool):	1✔
49	def decorator(f):	1✔
50	for d in reversed(	1✔
51	[
52	_opt_asana_token_pass_path,
53	_opt_asana_workspace_gid,
54	_opt_asana_workspace_name,
55	_opt_list_asana_workspaces,
56	],
57	):
58	f = d()(f)	1✔
59
60	# --asana-task-gid is used to ease development and debugging. It is not currently
61	# suitable for regular use.
62	return _opt_asana_task_gid(hidden=hidden_gid)(f)	1✔
63
64	return decorator	1✔
65
66
67	def _opt_asana_task_gid(**kwargs):	1✔
68	return click.option(	1✔
69	"-a",
70	"--asana-task-gid",
71	"asana_task_gid",
72	type=str,
73	help="Limit sync to provided task",
74	**kwargs,
75	)
76
77
78	def _opt_asana_token_pass_path():	1✔
79	def callback(ctx, param, value):	1✔
80	del ctx	×
81
82	api_token_pass_path = value	×
83
84	# fetch API token to connect to asana -------------------------------------------------
85	asana_token = os.environ.get("ASANA_PERSONAL_ACCESS_TOKEN")	×
86
87	if asana_token is None and api_token_pass_path is None:	×
88	error_and_exit(	×
89	"You must provide an Asana Personal Access asana_token, using the"
90	f" {'/'.join(param.opts)} option",
91	)
92	if asana_token is not None:	×
93	logger.debug(	×
94	"Reading the Asana Personal Access asana_token (PAT) from environment"
95	" variable...",
96	)
97	else:
98	asana_token = fetch_from_pass_manager(api_token_pass_path)	×
99
100	return asana_token	×
101
102	return click.option(	1✔
103	"--token",
104	"--token-pass-path",
105	"asana_token",
106	help="Path in the UNIX password manager to fetch",
107	expose_value=True,
108	callback=callback,
109	)
110
111
112	def _opt_asana_workspace_gid():	1✔
113	return click.option(	1✔
114	"-w",
115	"--asana-workspace-gid",
116	"asana_workspace_gid",
117	type=str,
118	help="Asana workspace GID used to filter tasks",
119	)
120
121
122	def _opt_asana_workspace_name():	1✔
123	return click.option(	1✔
124	"-W",
125	"--asana-workspace-name",
126	"asana_workspace_name",
127	type=str,
128	help="Asana workspace name used to filter tasks",
129	)
130
131
132	def _opt_list_asana_workspaces():	1✔
133	return click.option(	1✔
134	"--list-asana-workspaces",
135	"do_list_asana_workspaces",
136	is_flag=True,
137	help="List the available Asana workspaces",
138	)
139
140
141	# taskwarrior options -------------------------------------------------------------------------
142	def opts_tw_filtering():	1✔
143	def decorator(f):	1✔
144	for d in reversed(	1✔
145	[
146	_opt_tw_filter,
147	_opt_tw_all_tasks,
148	_opt_tw_tags,
149	_opt_tw_project,
150	_opt_tw_only_tasks_modified_X_days,
151	_opt_prefer_scheduled_date,
152	],
153	):
154	f = d()(f)	1✔
155	return f	1✔
156
157	return decorator	1✔
158
159
160	def _opt_tw_filter():	1✔
161	return click.option(	1✔
162	"-f",
163	"--tw-filter",
164	"tw_filter",
165	type=str,
166	help=(
167	"Taskwarrior filter for specifying the tasks to synchronize. These filters will be"
168	" concatenated using AND with potential tags and projects potentially specified."
169	),
170	)
171
172
173	def _opt_tw_all_tasks():	1✔
174	return click.option(	1✔
175	"--all",
176	"--taskwarrior-all-tasks",
177	"tw_sync_all_tasks",
178	is_flag=True,
179	help="Sync all taskwarrior tasks (potentially very slow).",
180	)
181
182
183	def _opt_tw_tags():	1✔
184	return click.option(	1✔
185	"-t",
186	"--taskwarrior-tags",
187	"tw_tags",
188	type=str,
189	help="Taskwarrior tags to synchronize.",
190	expose_value=True,
191	multiple=True,
192	)
193
194
195	def _opt_tw_project():	1✔
196	return click.option(	1✔
197	"-p",
198	"--tw-project",
199	"tw_project",
200	type=str,
201	help="Taskwarrior project to synchronize.",
202	expose_value=True,
203	is_eager=True,
204	)
205
206
207	def _opt_tw_only_tasks_modified_X_days():	1✔
208	def callback(ctx, param, value):	1✔
209	del param	×
210
211	if value is None or ctx.resilient_parsing:	×
212	return None	×
213
214	return f"modified.after:-{value}d"	×
215
216	return click.option(	1✔
217	"--days",
218	"--only-modified-last-X-days",
219	"tw_only_modified_last_X_days",
220	type=str,
221	help=(
222	"Only synchronize Taskwarrior tasks that have been modified in the last X days"
223	" (specify X, e.g., 1, 30, 0.5, etc.)."
224	),
225	callback=callback,
226	)
227
228
229	def _opt_prefer_scheduled_date():	1✔
230	return click.option(	1✔
231	"--prefer-scheduled-date",
232	"prefer_scheduled_date",
233	is_flag=True,
234	help=(
235	'Prefer using the "scheduled" date field instead of the "due" date if the former'
236	" is available."
237	),
238	)
239
240
241	# notion --------------------------------------------------------------------------------------
242	def opt_notion_page_id():	1✔
243	return click.option(	1✔
244	"-n",
245	"--notion-page",
246	"notion_page_id",
247	type=str,
248	help="Page ID of the Notion page to synchronize.",
249	)
250
251
252	def opt_notion_token_pass_path():	1✔
253	return click.option(	1✔
254	"--token",
255	"--token-pass-path",
256	"token_pass_path",
257	help="Path in the UNIX password manager to fetch.",
258	)
259
260
261	# gkeep ---------------------------------------------------------------------------------------
262	def opts_gkeep():	1✔
263	def decorator(f):	1✔
264	for d in reversed(	1✔
265	[
266	_opt_gkeep_user_pass_path,
267	_opt_gkeep_passwd_pass_path,
268	_opt_gkeep_token_pass_path,
269	],
270	):
271	f = d()(f)	1✔
272
273	return f	1✔
274
275	return decorator	1✔
276
277
278	def _opt_gkeep_user_pass_path():	1✔
279	return click.option(	1✔
280	"--user",
281	"--user-pass-path",
282	"gkeep_user_pass_path",
283	help="Path in the UNIX password manager to fetch the Google username from.",
284	default="gkeepapi/user",
285	)
286
287
288	def _opt_gkeep_passwd_pass_path():	1✔
289	return click.option(	1✔
290	"--passwd",
291	"--passwd-pass-path",
292	"gkeep_passwd_pass_path",
293	help="Path in the UNIX password manager to fetch the Google password from.",
294	default="gkeepapi/passwd",
295	)
296
297
298	def _opt_gkeep_token_pass_path():	1✔
299	return click.option(	1✔
300	"--token",
301	"--token-pass-path",
302	"gkeep_token_pass_path",
303	help="Path in the UNIX password manager to fetch the google keep token from.",
304	default="gkeepapi/token",
305	)
306
307
308	def opt_gkeep_labels():	1✔
309	return click.option(	1✔
310	"-k",
311	"--gkeep-labels",
312	type=str,
313	multiple=True,
314	help="Google Keep labels whose notes to synchronize.",
315	)
316
317
318	def opt_gkeep_ignore_labels():	1✔
319	return click.option(	1✔
320	"-i",
321	"--gkeep-ignore-labels",
322	type=str,
323	multiple=True,
324	help="Google Keep labels whose notes will be explicitly ignored.",
325	)
326
327
328	def opt_gkeep_note():	1✔
329	return click.option(	1✔
330	"-k",
331	"--gkeep-note",
332	type=str,
333	help=(
334	"Full title of the Google Keep Note to synchronize - Make sure you enable the"
335	" checkboxes."
336	),
337	)
338
339
340	# google calendar -----------------------------------------------------------------------------
341	def opt_gcal_calendar():	1✔
342	return click.option(	1✔
343	"-c",
344	"--gcal-calendar",
345	type=str,
346	help="Name of the Google Calendar to synchronize (will be created if not there).",
347	)
348
349
350	# google tasks --------------------------------------------------------------------------------
351	def opt_gtasks_list():	1✔
352	return click.option(	1✔
353	"-l",
354	"--gtasks-list",
355	type=str,
356	help="Name of the Google Tasks list to synchronize (will be created if not there).",
357	)
358
359
360	# markdown file -------------------------------------------------------------------------------
361	def opts_markdown():	1✔
362	def decorator(f):	1✔
363	for d in reversed(	1✔
364	[
365	_opt_md_file,
366	_opt_prefer_scheduled_date,
367	],
368	):
369	f = d()(f)	1✔
370	return f	1✔
371
372	return decorator	1✔
373
374
375	def _opt_md_file():	1✔
376	return click.option(	1✔
377	"-m",
378	"--markdown-file",
379	type=str,
380	help="Name of the Markdown file including tasks list to synchronize",
381	)
382
383
384	# google-related options ----------------------------------------------------------------------
385	def opt_google_secret_override():	1✔
386	return click.option(	1✔
387	"--google-secret",
388	default=None,
389	type=click.Path(exists=True, file_okay=True, dir_okay=False, resolve_path=True),
390	help="Override the client secret used for the communication with the Google APIs.",
391	)
392
393
394	def opt_google_oauth_port():	1✔
395	return click.option(	1✔
396	"--oauth-port",
397	default=8081,
398	type=int,
399	help="Port to use for OAuth Authentication with Google Applications.",
400	)
401
402
403	# caldav options ------------------------------------------------------------------------------
404	def opts_caldav():	1✔
405	def decorator(f):	1✔
406	for d in reversed(	1✔
407	[
408	_opt_caldav_calendar,
409	_opt_caldav_url,
410	_opt_caldav_user,
411	_opt_caldav_passwd_pass_path,
412	_opt_caldav_passwd_cmd,
413	],
414	):
415	f = d()(f)	1✔
416
417	return f	1✔
418
419	return decorator	1✔
420
421
422	def _opt_caldav_calendar():	1✔
423	return click.option(	1✔
424	"--caldav-calendar",
425	type=str,
426	default="Personal",
427	help="Name of the caldav Calendar to sync (will be created if not there).",
428	)
429
430
431	def _opt_caldav_url():	1✔
432	return click.option(	1✔
433	"--caldav-url",
434	type=str,
435	help="URL where the caldav calendar is hosted at (including /dav if applicable).",
436	)
437
438
439	def _opt_caldav_user():	1✔
440	return click.option(	1✔
441	"--caldav-user",
442	"caldav_user",
443	help="The caldav username for the given caldav instance",
444	)
445
446
447	def _opt_caldav_passwd_pass_path():	1✔
448	return click.option(	1✔
449	"--caldav-passwd",
450	"--caldav-passwd-pass-path",
451	"caldav_passwd_pass_path",
452	help="Path in the UNIX password manager to fetch the caldav password from.",
453	)
454
455
456	def _opt_caldav_passwd_cmd():	1✔
457	return click.option(	1✔
458	"--caldav-passwd-cmd",
459	"caldav_passwd_cmd",
460	help="Command that outputs the caldav password on stdout.",
461	)
462
463
464	# filesystem related options ------------------------------------------------------------------
465	def opt_filesystem_root():	1✔
466	return click.option(	1✔
467	"--fs",
468	"--fs-root",
469	"filesystem_root",
470	required=False,
471	type=str,
472	help="Directory to consider as root for synchronization operations.",
473	)
474
475
476	def opt_filename_extension():	1✔
477	return click.option(	1✔
478	"--ext",
479	"--filename-extension",
480	"filename_extension",
481	type=str,
482	help="Use this extension for locally created files.",
483	default=".md",
484	)
485
486	def opt_filename_path():	1✔
NEW 487	return click.option(	×
488	"--path",
489	"--filename-path",
490	"filename_path",
491	type=str,
492	help="Use this file path for locally saved data",
493	)
494
495	# general options -----------------------------------------------------------------------------
496	def opts_miscellaneous(side_A_name: str, side_B_name: str):	1✔
497	def decorator(f):	1✔
498	for d in reversed(	1✔
499	[
500	(_opt_list_resolution_strategies,),
501	(_opt_resolution_strategy,),
502	(_opt_confirm,),
503	(
504	click.version_option,
505	version,
506	),
507	(_opt_pdb_on_error,),
508	(_opt_list_combinations, side_A_name, side_B_name),
509	(_opt_combination, side_A_name, side_B_name),
510	(_opt_custom_combination_savename, side_A_name, side_B_name),
511	],
512	):
513	fn = d[0]	1✔
514	fn_args = d[1:]	1✔
515	f = fn(*fn_args)(f) # type: ignore	1✔
516
517	return click.option("-v", "--verbose", count=True)(f)	1✔
518
519	return decorator	1✔
520
521
522	def _opt_confirm():	1✔
523	return click.option(	1✔
524	"--confirm",
525	"confirm",
526	is_flag=True,
527	default=False,
528	expose_value=True,
529	help="Confirm app configuration before proceeding with the actual execution",
530	)
531
532
533	def opt_default_duration_event_mins():	1✔
534	return click.option(	1✔
535	"--default-event-duration-mins",
536	"default_event_duration_mins",
537	default=30,
538	type=int,
539	help="The default duration of an event that is to be created [in minutes].",
540	)
541
542
543	def _list_named_combinations(config_fname: str) -> None:	1✔
544	"""List the named configurations currently available for the given configuration name."""
545	logger.success(	1✔
546	format_list(
547	header="\n\nNamed configurations currently available",
548	items=get_named_combinations(config_fname=config_fname),
549	),
550	)
551
552
553	def _opt_list_combinations(side_A_name: str, side_B_name: str):	1✔
554	def callback(ctx, param, value):	1✔
555	del ctx, param	×
556	if value is True:	×
557	_list_named_combinations(	×
558	config_fname=determine_app_config_fname(
559	side_A_name=side_A_name,
560	side_B_name=side_B_name,
561	),
562	)
563	sys.exit(0)	×
564
565	return click.option(	1✔
566	"--list-combinations",
567	"do_list_combinations",
568	is_flag=True,
569	expose_value=False,
570	help=f"List the available named {side_A_name}<->{side_B_name} combinations.",
571	callback=callback,
572	)
573
574
575	def _opt_resolution_strategy():	1✔
576	return click.option(	1✔
577	"-r",
578	"--resolution-strategy",
579	default="AlwaysSecondRS",
580	type=click.Choice(list(name_to_resolution_strategy_type.keys())),
581	help="Resolution strategy to use during conflicts.",
582	)
583
584
585	def _opt_list_resolution_strategies():	1✔
586	def _list_resolution_strategies(ctx, param, value):	1✔
587	del ctx, param	×
588
589	if value is not True:	×
590	return	×
591
592	strs = name_to_resolution_strategy_type.keys()	×
593	click.echo(	×
594	"\n".join(
595	[f"{a}. {b}" for a, b in zip(range(1, len(strs) + 1), strs, strict=False)],
596	),
597	)
598	sys.exit(0)	×
599
600	return click.option(	1✔
601	"--list-resolution-strategies",
602	callback=_list_resolution_strategies,
603	is_flag=True,
604	expose_value=False,
605	help="List all the available resolution strategies and exit.",
606	)
607
608
609	def _opt_combination(side_A_name: str, side_B_name: str):	1✔
610	return click.option(	1✔
611	COMBINATION_FLAGS[0],
612	COMBINATION_FLAGS[1],
613	"combination_name",
614	type=str,
615	help=f"Name of an already saved {side_A_name}<->{side_B_name} combination.",
616	)
617
618
619	def _opt_custom_combination_savename(side_A_name: str, side_B_name: str):	1✔
620	return click.option(	1✔
621	"-s",
622	"--save-as",
623	"custom_combination_savename",
624	type=str,
625	help=(
626	f"Save the given {side_A_name}<->{side_B_name} filters combination using a"
627	" specified custom name."
628	),
629	)

bergercookie / syncall / 28399320195

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