4162446847

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/bergercookie/syncall/commit/<a class=hub.com/bergercookie/syncall/commit/<a class="double-link" href="https://git"><a class=hub.com/bergercookie/syncall/commit/<a class="double-link" href="https://git"><a class=hub.com/bergercookie/syncall/commit/<a class="double-link" href="https://git"><a class=hub.com/bergercookie/syncall/commit/<a class="double-link" href="https://git"><a class=hub.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e">8b67f4cab">&lt;a href=&quot;https://github.com/bergercookie/syncall/commit/</a><a class="double-link" href="https://github.com/bergercookie/syncall/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/bergercookie/syncall/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/bergercookie/syncall/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e">8b67f4cab</a><a href="https://github.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e">&lt;a href=&quot;https://github.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/bergercookie/syncall/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/bergercookie/syncall/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e&amp;quot;&amp;gt;8b67f4cab&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06dbb183ca5e&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/bergercookie/syncall/commit/8b67f4cabf670c9371f87458587b06db&amp;lt;/a&amp;gt;b183ca5e</a>&quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/bergercookie/syncall/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/bergercookie/syncall/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/bergercookie/syncall/commit/f3c104d0d... (continued)

Pull Request Pull Request #63: Add tw caldav sync

Run Details

333 of 333 new or added lines in 6 files covered. (100.0%)

1539 of 2715 relevant lines covered (56.69%)

0.57 hits per line

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

36.9

/syncall/scripts/tw_caldav_sync.py

import datetime
import os
from typing import List, Optional

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

from syncall import inform_about_app_extras
from syncall.app_utils import error_and_exit
from syncall.cli import (
    opt_caldav_calendar,
    opt_caldav_passwd_pass_path,
    opt_caldav_url,
    opt_caldav_user,
    opt_tw_all_tasks,
    opt_tw_only_tasks_modified_30_days,
)
from syncall.tw_caldav_utils import convert_caldav_to_tw, convert_tw_to_caldav

try:
    from syncall.caldav.caldav_side import CaldavSide
    from syncall.taskwarrior.taskwarrior_side import TaskWarriorSide
except ImportError:
    inform_about_app_extras(["caldav", "tw"])

from syncall import (
    Aggregator,
    __version__,
    cache_or_reuse_cached_combination,
    fetch_app_configuration,
    fetch_from_pass_manager,
    get_resolution_strategy,
    inform_about_combination_name_usage,
    list_named_combinations,
    opt_combination,
    opt_custom_combination_savename,
    opt_list_combinations,
    opt_resolution_strategy,
    opt_tw_project,
    opt_tw_tags,
    report_toplevel_exception,
)


@click.command()
# caldav options ---------------------------------------------------------------------
@opt_caldav_calendar()
@opt_caldav_url()
@opt_caldav_user()
@opt_caldav_passwd_pass_path()
# taskwarrior options -------------------------------------------------------------------------
@opt_tw_all_tasks()
@opt_tw_tags()
@opt_tw_project()
@opt_tw_only_tasks_modified_30_days()
# misc options --------------------------------------------------------------------------------
@opt_list_combinations("TW", "Caldav")
@opt_resolution_strategy()
@opt_combination("TW", "Caldav")
@opt_custom_combination_savename("TW", "Caldav")
@click.option("-v", "--verbose", count=True)
@click.version_option(__version__)
def main(
    caldav_calendar: str,
    caldav_url: str,
    caldav_user: Optional[str],
    caldav_passwd_pass_path: str,
    tw_sync_all_tasks: bool,
    tw_tags: List[str],
    tw_project: str,
    tw_only_modified_last_30_days: bool,
    verbose: int,
    combination_name: str,
    custom_combination_savename: str,
    resolution_strategy: str,
    do_list_combinations: bool,
):
    """Synchronize lists of tasks from your caldav Calendar with filters from Taskwarrior.

    The list of TW tasks is determined by a combination of TW tags and a TW project. Use
    `--all` to synchronize all tasks.

    The calendar in Caldav should be provided by their name. If it doesn't exist it will be
    created.
    """

    loguru_tqdm_sink(verbosity=verbose)
    log_to_syslog(name="tw_caldav_sync")
    logger.debug("Initialising...")
    inform_about_config = False

    if do_list_combinations:
        list_named_combinations(config_fname="tw_caldav_configs")
        return 0

    combination_of_tw_tags_and_tw_project = any([tw_tags, tw_project])
    check_required_mutually_exclusive(
        tw_sync_all_tasks,
        combination_of_tw_tags_and_tw_project,
        "All TW Tasks",
        "TW Tags + Projects",
    )

    check_optional_mutually_exclusive(combination_name, custom_combination_savename)
    combination_of_tw_project_tags_and_caldav_calendar = any(
        [
            tw_project,
            tw_tags,
            tw_sync_all_tasks,
            caldav_calendar,
        ]
    )
    check_optional_mutually_exclusive(
        combination_name, combination_of_tw_project_tags_and_caldav_calendar
    )

    # existing combination name is provided ---------------------------------------------------
    if combination_name is not None:
        app_config = fetch_app_configuration(
            config_fname="tw_caldav_configs", combination=combination_name
        )
        tw_tags = app_config["tw_tags"]
        tw_project = app_config["tw_project"]
        caldav_calendar = app_config["caldav_calendar"]

    # combination manually specified ----------------------------------------------------------
    else:
        inform_about_config = True
        combination_name = cache_or_reuse_cached_combination(
            config_args={
                "caldav_calendar": caldav_calendar,
                "tw_project": tw_project,
                "tw_tags": tw_tags,
                "tw_sync_all_tasks": tw_sync_all_tasks,
                "tw_only_modified_last_30_days": tw_only_modified_last_30_days,
            },
            config_fname="tw_caldav_configs",
            custom_combination_savename=custom_combination_savename,
        )

    # announce configuration ------------------------------------------------------------------
    logger.info(
        format_dict(
            header="Configuration",
            items={
                "TW Tags": tw_tags,
                "TW Project": tw_project,
                "TW Sync All Tasks": tw_sync_all_tasks,
                "Caldav Calendar": caldav_calendar,
            },
            prefix="\n\n",
            suffix="\n",
        )
    )

    # initialize sides ------------------------------------------------------------------------
    # TW
    # TODO abstract this
    only_modified_since = None
    if tw_only_modified_last_30_days:
        only_modified_since = datetime.datetime.now() - datetime.timedelta(days=30)

    tw_side = TaskWarriorSide(
        tags=tw_tags, project=tw_project, only_modified_since=only_modified_since
    )

    # caldav
    if not caldav_url or not caldav_calendar:
        logger.debug(caldav_url)
        logger.debug(caldav_calendar)
        error_and_exit(
            "You must provide a URL and calendar in order to synchronize via caldav"
        )

    # fetch username
    if not caldav_user:
        caldav_user = os.environ.get("CALDAV_USERNAME")
    if caldav_user is None:
        error_and_exit(
            "You must provide a username in order to synchronize via caldav, either "
        )

    # fetch password
    caldav_passwd = os.environ.get("CALDAV_PASSWD")
    if caldav_passwd is not None:
        logger.debug("Reading the caldav password from environment variable...")
    else:
        caldav_passwd = fetch_from_pass_manager(caldav_passwd_pass_path)

    client = caldav.DAVClient(url=caldav_url, username=caldav_user, password=caldav_passwd)
    caldav_side = CaldavSide(client=client, calendar_name=caldav_calendar)

    # sync ------------------------------------------------------------------------------------
    try:
        with Aggregator(
            side_A=caldav_side,
            side_B=tw_side,
            converter_B_to_A=convert_tw_to_caldav,
            converter_A_to_B=convert_caldav_to_tw,
            resolution_strategy=get_resolution_strategy(
                resolution_strategy, side_A_type=type(caldav_side), side_B_type=type(tw_side)
            ),
            config_fname=combination_name,
            ignore_keys=(
                (),
                (),
            ),
        ) as aggregator:
            aggregator.sync()
    except KeyboardInterrupt:
        logger.error("Exiting...")
        return 1
    except:
        report_toplevel_exception(is_verbose=verbose >= 1)
        return 1

    if inform_about_config:
        inform_about_combination_name_usage(combination_name)

    return 0


if __name__ == "__main__":
    main()

1	import datetime	1✔
2	import os	1✔
3	from typing import List, Optional	1✔
4
5	import caldav	1✔
6	import click	1✔
7	from bubop import (	1✔
8	check_optional_mutually_exclusive,
9	check_required_mutually_exclusive,
10	format_dict,
11	log_to_syslog,
12	logger,
13	loguru_tqdm_sink,
14	)
15
16	from syncall import inform_about_app_extras	1✔
17	from syncall.app_utils import error_and_exit	1✔
18	from syncall.cli import (	1✔
19	opt_caldav_calendar,
20	opt_caldav_passwd_pass_path,
21	opt_caldav_url,
22	opt_caldav_user,
23	opt_tw_all_tasks,
24	opt_tw_only_tasks_modified_30_days,
25	)
26	from syncall.tw_caldav_utils import convert_caldav_to_tw, convert_tw_to_caldav	1✔
27
28	try:	1✔
29	from syncall.caldav.caldav_side import CaldavSide	1✔
30	from syncall.taskwarrior.taskwarrior_side import TaskWarriorSide	1✔
31	except ImportError:	×
32	inform_about_app_extras(["caldav", "tw"])	×
33
34	from syncall import (	1✔
35	Aggregator,
36	__version__,
37	cache_or_reuse_cached_combination,
38	fetch_app_configuration,
39	fetch_from_pass_manager,
40	get_resolution_strategy,
41	inform_about_combination_name_usage,
42	list_named_combinations,
43	opt_combination,
44	opt_custom_combination_savename,
45	opt_list_combinations,
46	opt_resolution_strategy,
47	opt_tw_project,
48	opt_tw_tags,
49	report_toplevel_exception,
50	)
51
52
53	@click.command()	1✔
54	# caldav options ---------------------------------------------------------------------
55	@opt_caldav_calendar()	1✔
56	@opt_caldav_url()	1✔
57	@opt_caldav_user()	1✔
58	@opt_caldav_passwd_pass_path()	1✔
59	# taskwarrior options -------------------------------------------------------------------------
60	@opt_tw_all_tasks()	1✔
61	@opt_tw_tags()	1✔
62	@opt_tw_project()	1✔
63	@opt_tw_only_tasks_modified_30_days()	1✔
64	# misc options --------------------------------------------------------------------------------
65	@opt_list_combinations("TW", "Caldav")	1✔
66	@opt_resolution_strategy()	1✔
67	@opt_combination("TW", "Caldav")	1✔
68	@opt_custom_combination_savename("TW", "Caldav")	1✔
69	@click.option("-v", "--verbose", count=True)	1✔
70	@click.version_option(__version__)	1✔
71	def main(	1✔
72	caldav_calendar: str,
73	caldav_url: str,
74	caldav_user: Optional[str],
75	caldav_passwd_pass_path: str,
76	tw_sync_all_tasks: bool,
77	tw_tags: List[str],
78	tw_project: str,
79	tw_only_modified_last_30_days: bool,
80	verbose: int,
81	combination_name: str,
82	custom_combination_savename: str,
83	resolution_strategy: str,
84	do_list_combinations: bool,
85	):
86	"""Synchronize lists of tasks from your caldav Calendar with filters from Taskwarrior.
87
88	The list of TW tasks is determined by a combination of TW tags and a TW project. Use
89	`--all` to synchronize all tasks.
90
91	The calendar in Caldav should be provided by their name. If it doesn't exist it will be
92	created.
93	"""
94
95	loguru_tqdm_sink(verbosity=verbose)	×
96	log_to_syslog(name="tw_caldav_sync")	×
97	logger.debug("Initialising...")	×
98	inform_about_config = False	×
99
100	if do_list_combinations:	×
101	list_named_combinations(config_fname="tw_caldav_configs")	×
102	return 0	×
103
104	combination_of_tw_tags_and_tw_project = any([tw_tags, tw_project])	×
105	check_required_mutually_exclusive(	×
106	tw_sync_all_tasks,
107	combination_of_tw_tags_and_tw_project,
108	"All TW Tasks",
109	"TW Tags + Projects",
110	)
111
112	check_optional_mutually_exclusive(combination_name, custom_combination_savename)	×
113	combination_of_tw_project_tags_and_caldav_calendar = any(	×
114	[
115	tw_project,
116	tw_tags,
117	tw_sync_all_tasks,
118	caldav_calendar,
119	]
120	)
121	check_optional_mutually_exclusive(	×
122	combination_name, combination_of_tw_project_tags_and_caldav_calendar
123	)
124
125	# existing combination name is provided ---------------------------------------------------
126	if combination_name is not None:	×
127	app_config = fetch_app_configuration(	×
128	config_fname="tw_caldav_configs", combination=combination_name
129	)
130	tw_tags = app_config["tw_tags"]	×
131	tw_project = app_config["tw_project"]	×
132	caldav_calendar = app_config["caldav_calendar"]	×
133
134	# combination manually specified ----------------------------------------------------------
135	else:
136	inform_about_config = True	×
137	combination_name = cache_or_reuse_cached_combination(	×
138	config_args={
139	"caldav_calendar": caldav_calendar,
140	"tw_project": tw_project,
141	"tw_tags": tw_tags,
142	"tw_sync_all_tasks": tw_sync_all_tasks,
143	"tw_only_modified_last_30_days": tw_only_modified_last_30_days,
144	},
145	config_fname="tw_caldav_configs",
146	custom_combination_savename=custom_combination_savename,
147	)
148
149	# announce configuration ------------------------------------------------------------------
150	logger.info(	×
151	format_dict(
152	header="Configuration",
153	items={
154	"TW Tags": tw_tags,
155	"TW Project": tw_project,
156	"TW Sync All Tasks": tw_sync_all_tasks,
157	"Caldav Calendar": caldav_calendar,
158	},
159	prefix="\n\n",
160	suffix="\n",
161	)
162	)
163
164	# initialize sides ------------------------------------------------------------------------
165	# TW
166	# TODO abstract this
167	only_modified_since = None	×
168	if tw_only_modified_last_30_days:	×
169	only_modified_since = datetime.datetime.now() - datetime.timedelta(days=30)	×
170
171	tw_side = TaskWarriorSide(	×
172	tags=tw_tags, project=tw_project, only_modified_since=only_modified_since
173	)
174
175	# caldav
176	if not caldav_url or not caldav_calendar:	×
177	logger.debug(caldav_url)	×
178	logger.debug(caldav_calendar)	×
179	error_and_exit(	×
180	"You must provide a URL and calendar in order to synchronize via caldav"
181	)
182
183	# fetch username
184	if not caldav_user:	×
185	caldav_user = os.environ.get("CALDAV_USERNAME")	×
186	if caldav_user is None:	×
187	error_and_exit(	×
188	"You must provide a username in order to synchronize via caldav, either "
189	)
190
191	# fetch password
192	caldav_passwd = os.environ.get("CALDAV_PASSWD")	×
193	if caldav_passwd is not None:	×
194	logger.debug("Reading the caldav password from environment variable...")	×
195	else:
196	caldav_passwd = fetch_from_pass_manager(caldav_passwd_pass_path)	×
197
198	client = caldav.DAVClient(url=caldav_url, username=caldav_user, password=caldav_passwd)	×
199	caldav_side = CaldavSide(client=client, calendar_name=caldav_calendar)	×
200
201	# sync ------------------------------------------------------------------------------------
202	try:	×
203	with Aggregator(	×
204	side_A=caldav_side,
205	side_B=tw_side,
206	converter_B_to_A=convert_tw_to_caldav,
207	converter_A_to_B=convert_caldav_to_tw,
208	resolution_strategy=get_resolution_strategy(
209	resolution_strategy, side_A_type=type(caldav_side), side_B_type=type(tw_side)
210	),
211	config_fname=combination_name,
212	ignore_keys=(
213	(),
214	(),
215	),
216	) as aggregator:
217	aggregator.sync()	×
218	except KeyboardInterrupt:	×
219	logger.error("Exiting...")	×
220	return 1	×
221	except:	×
222	report_toplevel_exception(is_verbose=verbose >= 1)	×
223	return 1	×
224
225	if inform_about_config:	×
226	inform_about_combination_name_usage(combination_name)	×
227
228	return 0	×
229
230
231	if __name__ == "__main__":	1✔
232	main()	×

bergercookie / syncall / 4162446847

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