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

36.54

/syncall/scripts/md_gtasks_sync.py

from typing import List

import click
from bubop import (
    check_optional_mutually_exclusive,
    check_required_mutually_exclusive,
    format_dict,
    logger,
    loguru_tqdm_sink,
)

from syncall.app_utils import confirm_before_proceeding, inform_about_app_extras

try:
    from syncall.google.gtasks_side import GTasksSide
    from syncall.filesystem.markdown_tasks_side import MarkdownTasksSide
except ImportError:
    inform_about_app_extras(["google", "fs"])

from syncall.aggregator import Aggregator
from syncall.app_utils import (
    app_log_to_syslog,
    cache_or_reuse_cached_combination,
    error_and_exit,
    fetch_app_configuration,
    get_resolution_strategy,
    register_teardown_handler,
)
from syncall.cli import (
    opt_google_oauth_port,
    opt_google_secret_override,
    opt_gtasks_list,
    opts_markdown,
    opts_miscellaneous,
)
from syncall.tw_gtasks_utils import convert_gtask_to_md, convert_md_to_gtask


@click.command()
@opt_gtasks_list()
@opt_google_secret_override()
@opt_google_oauth_port()
@opts_markdown()
@opts_miscellaneous(side_A_name="Obsidian", side_B_name="Google Tasks")
def main(
    gtasks_list: str,
    google_secret: str,
    oauth_port: int,
    markdown_file: str,
    prefer_scheduled_date: bool,
    resolution_strategy: str,
    verbose: int,
    combination_name: str,
    custom_combination_savename: str,
    pdb_on_error: bool,
    confirm: bool,
):
    """Synchronize lists from your Google Tasks with Obsidian Tasks Markdown file.

    The list of MD tasks can be based on a Markdown file path
    while the list in GTasks should be provided by their name. if it doesn't
    exist it will be created.
    """
    # setup logger ----------------------------------------------------------------------------
    loguru_tqdm_sink(verbosity=verbose)
    app_log_to_syslog()
    logger.debug("Initialising...")
    inform_about_config = False

    # cli validation --------------------------------------------------------------------------
    check_optional_mutually_exclusive(combination_name, custom_combination_savename)

    combination_of_file_and_gtasks_list = any(
        [
            markdown_file,
            gtasks_list,
        ]
    )
    check_optional_mutually_exclusive(
        combination_name, combination_of_file_and_gtasks_list
    )

    # existing combination name is provided ---------------------------------------------------
    if combination_name is not None:
        app_config = fetch_app_configuration(
            side_A_name="Obsidian", side_B_name="Google Tasks", combination=combination_name
        )
        markdown_file = app_config["markdown_file"]
        gtasks_list = app_config["gtasks_list"]

    # combination manually specified ----------------------------------------------------------
    else:
        inform_about_config = True
        combination_name = cache_or_reuse_cached_combination(
            config_args={
                "gtasks_list": gtasks_list,
                "markdown_file": markdown_file,
            },
            config_fname="md_gtasks_configs",
            custom_combination_savename=custom_combination_savename,
        )

    # more checks -----------------------------------------------------------------------------
    if gtasks_list is None:
        error_and_exit(
            "You have to provide the name of a Google Tasks list to synchronize events"
            " to/from. You can do so either via CLI arguments or by specifying an existing"
            " saved combination"
        )

    # announce configuration ------------------------------------------------------------------
    logger.info(
        format_dict(
            header="Configuration",
            items={
                "Markdown Filename Path": markdown_file,
                "Google Tasks": gtasks_list,
                "Prefer scheduled dates": prefer_scheduled_date,
            },
            prefix="\n\n",
            suffix="\n",
        )
    )
    if confirm:
        confirm_before_proceeding()

    # initialize sides ------------------------------------------------------------------------
    md_side = MarkdownTasksSide(
        markdown_file=markdown_file
    )

    gtasks_side = GTasksSide(
        task_list_title=gtasks_list, oauth_port=oauth_port, client_secret=google_secret
    )

    # teardown function and exception handling ------------------------------------------------
    register_teardown_handler(
        pdb_on_error=pdb_on_error,
        inform_about_config=inform_about_config,
        combination_name=combination_name,
        verbose=verbose,
    )

    # take extra arguments into account -------------------------------------------------------
    def convert_B_to_A(*args, **kargs):
        return convert_md_to_gtask(
            *args,
            **kargs,
            set_scheduled_date=prefer_scheduled_date,
        )

    convert_B_to_A.__doc__ = convert_md_to_gtask.__doc__

    def convert_A_to_B(*args, **kargs):
        return convert_gtask_to_md(
            *args,
            **kargs,
            set_scheduled_date=prefer_scheduled_date,
        )

    convert_A_to_B.__doc__ = convert_gtask_to_md.__doc__

    # sync ------------------------------------------------------------------------------------
    with Aggregator(
        side_A=gtasks_side,
        side_B=md_side,
        converter_B_to_A=convert_B_to_A,
        converter_A_to_B=convert_A_to_B,
        resolution_strategy=get_resolution_strategy(
            resolution_strategy, side_A_type=type(gtasks_side), side_B_type=type(md_side)
        ),
        config_fname=combination_name,
        ignore_keys=(
            ("last_modified_date"),
            (),
        ),
    ) as aggregator:
        aggregator.sync()

    return 0


if __name__ == "__main__":
    main()

1	from typing import List	1✔
2
3	import click	1✔
4	from bubop import (	1✔
5	check_optional_mutually_exclusive,
6	check_required_mutually_exclusive,
7	format_dict,
8	logger,
9	loguru_tqdm_sink,
10	)
11
12	from syncall.app_utils import confirm_before_proceeding, inform_about_app_extras	1✔
13
14	try:	1✔
15	from syncall.google.gtasks_side import GTasksSide	1✔
16	from syncall.filesystem.markdown_tasks_side import MarkdownTasksSide	1✔
NEW 17	except ImportError:	×
NEW 18	inform_about_app_extras(["google", "fs"])	×
19
20	from syncall.aggregator import Aggregator	1✔
21	from syncall.app_utils import (	1✔
22	app_log_to_syslog,
23	cache_or_reuse_cached_combination,
24	error_and_exit,
25	fetch_app_configuration,
26	get_resolution_strategy,
27	register_teardown_handler,
28	)
29	from syncall.cli import (	1✔
30	opt_google_oauth_port,
31	opt_google_secret_override,
32	opt_gtasks_list,
33	opts_markdown,
34	opts_miscellaneous,
35	)
36	from syncall.tw_gtasks_utils import convert_gtask_to_md, convert_md_to_gtask	1✔
37
38
39	@click.command()	1✔
40	@opt_gtasks_list()	1✔
41	@opt_google_secret_override()	1✔
42	@opt_google_oauth_port()	1✔
43	@opts_markdown()	1✔
44	@opts_miscellaneous(side_A_name="Obsidian", side_B_name="Google Tasks")	1✔
45	def main(	1✔
46	gtasks_list: str,
47	google_secret: str,
48	oauth_port: int,
49	markdown_file: str,
50	prefer_scheduled_date: bool,
51	resolution_strategy: str,
52	verbose: int,
53	combination_name: str,
54	custom_combination_savename: str,
55	pdb_on_error: bool,
56	confirm: bool,
57	):
58	"""Synchronize lists from your Google Tasks with Obsidian Tasks Markdown file.
59
60	The list of MD tasks can be based on a Markdown file path
61	while the list in GTasks should be provided by their name. if it doesn't
62	exist it will be created.
63	"""
64	# setup logger ----------------------------------------------------------------------------
NEW 65	loguru_tqdm_sink(verbosity=verbose)	×
NEW 66	app_log_to_syslog()	×
NEW 67	logger.debug("Initialising...")	×
NEW 68	inform_about_config = False	×
69
70	# cli validation --------------------------------------------------------------------------
NEW 71	check_optional_mutually_exclusive(combination_name, custom_combination_savename)	×
72
NEW 73	combination_of_file_and_gtasks_list = any(	×
74	[
75	markdown_file,
76	gtasks_list,
77	]
78	)
NEW 79	check_optional_mutually_exclusive(	×
80	combination_name, combination_of_file_and_gtasks_list
81	)
82
83	# existing combination name is provided ---------------------------------------------------
NEW 84	if combination_name is not None:	×
NEW 85	app_config = fetch_app_configuration(	×
86	side_A_name="Obsidian", side_B_name="Google Tasks", combination=combination_name
87	)
NEW 88	markdown_file = app_config["markdown_file"]	×
NEW 89	gtasks_list = app_config["gtasks_list"]	×
90
91	# combination manually specified ----------------------------------------------------------
92	else:
NEW 93	inform_about_config = True	×
NEW 94	combination_name = cache_or_reuse_cached_combination(	×
95	config_args={
96	"gtasks_list": gtasks_list,
97	"markdown_file": markdown_file,
98	},
99	config_fname="md_gtasks_configs",
100	custom_combination_savename=custom_combination_savename,
101	)
102
103	# more checks -----------------------------------------------------------------------------
NEW 104	if gtasks_list is None:	×
NEW 105	error_and_exit(	×
106	"You have to provide the name of a Google Tasks list to synchronize events"
107	" to/from. You can do so either via CLI arguments or by specifying an existing"
108	" saved combination"
109	)
110
111	# announce configuration ------------------------------------------------------------------
NEW 112	logger.info(	×
113	format_dict(
114	header="Configuration",
115	items={
116	"Markdown Filename Path": markdown_file,
117	"Google Tasks": gtasks_list,
118	"Prefer scheduled dates": prefer_scheduled_date,
119	},
120	prefix="\n\n",
121	suffix="\n",
122	)
123	)
NEW 124	if confirm:	×
NEW 125	confirm_before_proceeding()	×
126
127	# initialize sides ------------------------------------------------------------------------
NEW 128	md_side = MarkdownTasksSide(	×
129	markdown_file=markdown_file
130	)
131
NEW 132	gtasks_side = GTasksSide(	×
133	task_list_title=gtasks_list, oauth_port=oauth_port, client_secret=google_secret
134	)
135
136	# teardown function and exception handling ------------------------------------------------
NEW 137	register_teardown_handler(	×
138	pdb_on_error=pdb_on_error,
139	inform_about_config=inform_about_config,
140	combination_name=combination_name,
141	verbose=verbose,
142	)
143
144	# take extra arguments into account -------------------------------------------------------
NEW 145	def convert_B_to_A(args, *kargs):	×
NEW 146	return convert_md_to_gtask(	×
147	*args,
148	**kargs,
149	set_scheduled_date=prefer_scheduled_date,
150	)
151
NEW 152	convert_B_to_A.__doc__ = convert_md_to_gtask.__doc__	×
153
NEW 154	def convert_A_to_B(args, *kargs):	×
NEW 155	return convert_gtask_to_md(	×
156	*args,
157	**kargs,
158	set_scheduled_date=prefer_scheduled_date,
159	)
160
NEW 161	convert_A_to_B.__doc__ = convert_gtask_to_md.__doc__	×
162
163	# sync ------------------------------------------------------------------------------------
NEW 164	with Aggregator(	×
165	side_A=gtasks_side,
166	side_B=md_side,
167	converter_B_to_A=convert_B_to_A,
168	converter_A_to_B=convert_A_to_B,
169	resolution_strategy=get_resolution_strategy(
170	resolution_strategy, side_A_type=type(gtasks_side), side_B_type=type(md_side)
171	),
172	config_fname=combination_name,
173	ignore_keys=(
174	("last_modified_date"),
175	(),
176	),
177	) as aggregator:
NEW 178	aggregator.sync()	×
179
NEW 180	return 0	×
181
182
183	if __name__ == "__main__":	1✔
NEW 184	main()	×

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