Some tidy up and extend use cases for new encrpytion checker.

Author: JoeZiminski

datashuttle/configs/rclone_configs.py

@@ -12,7 +12,7 @@
 import yaml
 
 from datashuttle.configs import canonical_folders
-from datashuttle.utils import utils
+from datashuttle.utils import rclone_encryption, utils
 
 
 class RCloneConfigs:
@@ -51,11 +51,9 @@ def load_rclone_config_is_encrypted(self) -> dict:
         called a lot, we track this explicitly when a rclone config is
         encrypted / unencrypted and store to disk between sessions.
         """
-        assert self.datashuttle_configs["connection_method"] in [
-            "ssh",
-            "aws",
-            "gdrive",
-        ]
+        assert rclone_encryption.connection_method_requires_encryption(
+            self.datashuttle_configs["connection_method"]
+        )
 
         if self.rclone_encryption_state_file_path.is_file():
             with open(self.rclone_encryption_state_file_path, "r") as file:
@@ -78,11 +76,9 @@ def set_rclone_config_encryption_state(self, value: bool) -> None:
         Note that this is stored to disk each call (rather than tracked locally) to ensure
         it is updated live if updated through the Python API while the TUI is also running.
         """
-        assert self.datashuttle_configs["connection_method"] in [
-            "ssh",
-            "aws",
-            "gdrive",
-        ]
+        assert rclone_encryption.connection_method_requires_encryption(
+            self.datashuttle_configs["connection_method"]
+        )
 
         rclone_config_is_encrypted = self.load_rclone_config_is_encrypted()
 
@@ -97,11 +93,9 @@ def get_rclone_config_encryption_state(
         self,
     ) -> dict:
         """Return whether the config file associated with the current `connection_method`."""
-        assert self.datashuttle_configs["connection_method"] in [
-            "ssh",
-            "aws",
-            "gdrive",
-        ]
+        assert rclone_encryption.connection_method_requires_encryption(
+            self.datashuttle_configs["connection_method"]
+        )
 
         rclone_config_is_encrypted = self.load_rclone_config_is_encrypted()
 

datashuttle/utils/rclone.py

@@ -115,7 +115,9 @@ def call_rclone_through_script_for_central_connection(
             shell=False,
         )
 
-        if cfg["connection_method"] in ["ssh", "gdrive", "aws"]:
+        if rclone_encryption.connection_method_requires_encryption(
+            cfg["connection_method"]
+        ):
             output = run_function_that_requires_encrypted_rclone_config_access(
                 cfg, lambda_func
             )
@@ -553,7 +555,9 @@ def check_successful_connection_and_raise_error_on_fail(cfg: Configs) -> None:
 
 def get_rclone_config_filepath(cfg: Configs) -> Path:
     """Get the path to the central Rclone config for the current `connection_method`."""
-    if cfg["connection_method"] in ["aws", "ssh", "gdrive"]:
+    if rclone_encryption.connection_method_requires_encryption(
+        cfg["connection_method"]
+    ):
         config_filepath = (
             cfg.rclone.get_rclone_central_connection_config_filepath()
         )
@@ -656,23 +660,17 @@ def transfer_data(
             cfg,
             f"{rclone_args('copy')} "
             f'"{local_filepath}" "{cfg.rclone.get_rclone_config_name()}:'
-            f'{central_filepath}" {extra_arguments} {get_config_arg(cfg)} --ask-password=false',  # TODO: handle the error
+            f'{central_filepath}" {extra_arguments} {get_config_arg(cfg)} --ask-password=false',
         )
 
     elif upload_or_download == "download":
         output = call_rclone_through_script_for_central_connection(
             cfg,
             f"{rclone_args('copy')} "
             f'"{cfg.rclone.get_rclone_config_name()}:'
-            f'{central_filepath}" "{local_filepath}"  {extra_arguments} {get_config_arg(cfg)} --ask-password=false',  # TODO: handle the error
+            f'{central_filepath}" "{local_filepath}"  {extra_arguments} {get_config_arg(cfg)} --ask-password=false',
         )
 
-    if (
-        cfg["connection_method"] in ["ssh", "aws", "gdrive"]
-        and cfg.rclone.get_rclone_config_encryption_state()
-    ):  # TODO: this is a quick and dirty fix but this MUST be handled better
-        rclone_encryption.remove_credentials_as_password_command()
-
     # 1) now 'for central connection' terminology is confused, one is for all and the other checks internally if it is aws or not. This is okay but must be consistent
     # 2) make a utils function to do the connection method check, this is still kind of weird / error prone