Enable virtual tags and refactor scripts

Author: bergercookie

syncall/cli.py

@@ -4,12 +4,18 @@
 extra dependency.
 """
 
+import os
 import sys
 
 import click
+from bubop import logger
 
 from syncall import __version__
-from syncall.app_utils import name_to_resolution_strategy_type
+from syncall.app_utils import (
+    error_and_exit,
+    fetch_from_pass_manager,
+    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
 
@@ -35,6 +41,27 @@ def opt_pdb_on_error():
 
 
 # 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.
+        f = opt_asana_task_gid(hidden=hidden_gid)(f)
+
+        return f
+
+    return decorator
+
+
 def opt_asana_task_gid(**kwargs):
     return click.option(
         "-a",
@@ -47,11 +74,34 @@ def opt_asana_task_gid(**kwargs):
 
 
 def opt_asana_token_pass_path():
+    def callback(ctx, param, value):
+        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",
-        "token_pass_path",
+        "asana_token",
         help="Path in the UNIX password manager to fetch",
+        expose_value=True,
+        callback=callback,
     )
 
 
@@ -305,6 +355,24 @@ def opt_google_oauth_port():
 
 
 # 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",
@@ -365,7 +433,7 @@ def opt_filename_extension():
         "--filename-extension",
         "filename_extension",
         type=str,
-        help=f"Use this extension for locally created files",
+        help="Use this extension for locally created files",
         default=".md",
     )
 

syncall/scripts/fs_gkeep_sync.py

@@ -6,7 +6,7 @@
 from bubop import check_optional_mutually_exclusive, format_dict, logger, loguru_tqdm_sink
 
 from syncall import inform_about_app_extras
-from syncall.cli import opt_filename_extension, opt_gkeep_ignore_labels
+from syncall.cli import opt_filename_extension, opt_gkeep_ignore_labels, opts_miscellaneous
 
 try:
     from syncall import GKeepNoteSide
@@ -16,31 +16,25 @@
 from syncall import (
     Aggregator,
     FilesystemSide,
-    __version__,
     cache_or_reuse_cached_combination,
     convert_filesystem_file_to_gkeep_note,
     convert_gkeep_note_to_filesystem_file,
     fetch_app_configuration,
     get_resolution_strategy,
-    inform_about_combination_name_usage,
     list_named_combinations,
-    report_toplevel_exception,
 )
 from syncall.app_utils import (
     app_log_to_syslog,
     gkeep_read_username_password_token,
+    register_teardown_handler,
     write_to_pass_manager,
 )
 from syncall.cli import (
-    opt_combination,
-    opt_custom_combination_savename,
     opt_filesystem_root,
     opt_gkeep_labels,
     opt_gkeep_passwd_pass_path,
     opt_gkeep_token_pass_path,
     opt_gkeep_user_pass_path,
-    opt_list_combinations,
-    opt_resolution_strategy,
 )
 
 
@@ -55,14 +49,10 @@
 @opt_filename_extension()
 @opt_filesystem_root()
 # misc options --------------------------------------------------------------------------------
-@opt_list_combinations("Filesystem", "Google Keep")
-@opt_resolution_strategy()
-@opt_combination("Filesystem", "Google Keep")
-@opt_custom_combination_savename("Filesystem", "Google Keep")
-@click.option("-v", "--verbose", count=True)
-@click.version_option(__version__)
+@opts_miscellaneous("Filesystem", "Google Keep")
 def main(
     filesystem_root: Optional[str],
+    filename_extension: str,
     gkeep_labels: Sequence[str],
     gkeep_ignore_labels: Sequence[str],
     gkeep_user_pass_path: str,
@@ -73,7 +63,8 @@ def main(
     combination_name: str,
     custom_combination_savename: str,
     do_list_combinations: bool,
-    filename_extension: str,
+    list_resolution_strategies: bool,
+    pdb_on_error: bool,
 ):
     """
     Synchronize Notes from your Google Keep with text files in a directory on your filesystem.
@@ -136,6 +127,7 @@ def main(
         gkeep_labels = app_config["gkeep_labels"]
         gkeep_ignore_labels = app_config["gkeep_ignore_labels"]
         filename_extension = app_config["filename_extension"]
+
     # combination manually specified ----------------------------------------------------------
     else:
         inform_about_config = True
@@ -188,7 +180,6 @@ def main(
         gkeep_token_pass_path,
     )
 
-    # initialize google keep  -----------------------------------------------------------------
     gkeep_side = GKeepNoteSide(
         gkeep_labels=gkeep_labels,
         gkeep_ignore_labels=gkeep_ignore_labels,
@@ -197,12 +188,19 @@ def main(
         gkeep_token=gkeep_token,
     )
 
-    # initialize Filesystem Side --------------------------------------------------------------
     filesystem_side = FilesystemSide(
         filesystem_root=filesystem_root_path, filename_extension=filename_extension
     )
 
-    # sync ------------------------------------------------------------------------------------
+    # 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 converter_A_to_B(gkeep_note):
         return convert_gkeep_note_to_filesystem_file(
             gkeep_note=gkeep_note,
@@ -212,40 +210,31 @@ def converter_A_to_B(gkeep_note):
 
     converter_A_to_B.__doc__ = convert_gkeep_note_to_filesystem_file.__doc__
 
-    try:
-        with Aggregator(
-            side_A=gkeep_side,
-            side_B=filesystem_side,
-            converter_B_to_A=convert_filesystem_file_to_gkeep_note,
-            converter_A_to_B=converter_A_to_B,
-            resolution_strategy=get_resolution_strategy(
-                resolution_strategy,
-                side_A_type=type(gkeep_side),
-                side_B_type=type(filesystem_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
-
-    # cache the token
+    # sync ------------------------------------------------------------------------------------
+    with Aggregator(
+        side_A=gkeep_side,
+        side_B=filesystem_side,
+        converter_B_to_A=convert_filesystem_file_to_gkeep_note,
+        converter_A_to_B=converter_A_to_B,
+        resolution_strategy=get_resolution_strategy(
+            resolution_strategy,
+            side_A_type=type(gkeep_side),
+            side_B_type=type(filesystem_side),
+        ),
+        config_fname=combination_name,
+        ignore_keys=(
+            (),
+            (),
+        ),
+    ) as aggregator:
+        aggregator.sync()
+
+    # cache the token -------------------------------------------------------------------------
     token = gkeep_side.get_master_token()
     if token is not None:
         logger.debug(f"Caching the gkeep token in pass -> {gkeep_token_pass_path}...")
         write_to_pass_manager(password_path=gkeep_token_pass_path, passwd=token)
 
-    if inform_about_config:
-        inform_about_combination_name_usage(combination_name)
-
     return 0