list branches sync action

Author: soustruh

component_config/configSchema.json

@@ -99,8 +99,17 @@
         },
         "branch": {
           "type": "string",
+          "enum": [],
           "title": "Branch Name",
-          "propertyOrder": 70
+          "propertyOrder": 70,
+          "options": {
+            "async": {
+              "cache": false,
+              "label": "List Branches",
+              "action": "listBranches",
+              "autoload": []
+            }
+          }
         },
         "filename": {
           "type": "string",

src/component.py

@@ -12,12 +12,12 @@
 from traceback import TracebackException
 
 import dacite
-from keboola.component.base import ComponentBase
+from keboola.component.base import ComponentBase, sync_action
 from keboola.component.exceptions import UserException
 
 import source_file
 import source_git
-from configuration import Configuration, SourceEnum, encrypted_keys
+from configuration import AuthEnum, Configuration, SourceEnum, encrypted_keys
 
 
 class Component(ComponentBase):
@@ -37,7 +37,7 @@ def __init__(self):
         self.parameters = dacite.from_dict(
             Configuration,
             self.configuration.parameters,
-            config=dacite.Config(cast=[SourceEnum], convert_key=encrypted_keys),
+            config=dacite.Config(cast=[AuthEnum, SourceEnum], convert_key=encrypted_keys),
         )
 
     def run(self):
@@ -122,6 +122,15 @@ def _merge_user_parameters(self):
         with open(os.path.join(self.data_folder_path, "config.json"), "w+") as inp:
             json.dump(config_data, inp)
 
+    @sync_action("listBranches")
+    def get_repository_branches(self):
+        """
+        Returns a list of branches in the git repository.
+        This method is used to populate the branches dropdown in the UI.
+        """
+        git_handler = source_git.GitHandler(self.parameters.git)
+        return git_handler.get_repository_branches()
+
 
 """
 Main entrypoint

src/source_git.py

@@ -6,7 +6,7 @@
 
 from keboola.component.exceptions import UserException
 
-from configuration import GitConfiguration
+from configuration import AuthEnum, GitConfiguration
 
 
 class GitHandler:
@@ -16,7 +16,65 @@ def __init__(self, git_cfg: GitConfiguration):
         # add path for absolute imports to start at the cloned repository root level
         sys.path.append(os.path.join(pathlib.Path(__file__).parent.parent, self.REPO_PATH))
 
+        self.env = os.environ.copy()
         self.git_cfg = git_cfg
+        self.repo_auth_url = None  # ‼️ NEVER EVER INCLUDE THIS VARIABLE IN LOGGING OUTPUT ‼️
+
+        if not self.git_cfg.url:
+            raise UserException("Git repository URL is required")
+
+        if self.git_cfg.auth == AuthEnum.PAT:
+            self._set_up_token_auth()
+        else:
+            self._set_up_ssh_command()
+
+        # do not ask for credentials when git authentication fails
+        self.env["GIT_TERMINAL_PROMPT"] = "0"
+
+    def _set_up_token_auth(self) -> None:
+        if not self.git_cfg.encrypted_token:
+            raise UserException("No personal access token provided")
+
+        if not self.git_cfg.url.startswith("https://"):
+            raise UserException("PAT authentication is only supported for HTTPS URLs")
+
+        self.repo_auth_url = self.git_cfg.url.replace(
+            "https://", f"https://x-token-auth:{self.git_cfg.encrypted_token}@"
+        )
+        logging.info("Git token authentication set up for HTTPS URL.")
+
+    def _set_up_ssh_command(self) -> None:
+        if self.git_cfg.auth == AuthEnum.SSH and not self.git_cfg.ssh_keys.keys.encrypted_private:
+            raise UserException("SSH key is required for SSH authentication")
+
+        repo_url = self.git_cfg.url
+        if repo_url.startswith("git@") or repo_url.startswith("ssh://"):
+            logging.warning("SSH URL detected but no ssh_key_path provided. Trying default SSH configuration.")
+
+        ssh_command = [
+            "ssh",
+            # the following lines could be used to disable strict host key checking, but it is better
+            # for security reasons to use the known_hosts file prepared in Dockerfile
+            # "-o",
+            # "StrictHostKeyChecking=no",
+            "-o",
+            "BatchMode=yes",  # do not ask for credentials when SSH auth fails
+            "-o",
+            "ConnectTimeout=30",
+            "-o",
+            "ServerAliveInterval=60",
+        ]
+
+        if self.git_cfg.ssh_keys.keys.encrypted_private:
+            ssh_key_path = os.path.expanduser("~/.ssh/github_private_key")
+            with open(ssh_key_path, "wb") as f:
+                for line in self.git_cfg.ssh_keys.keys.encrypted_private.splitlines():
+                    f.write(line.encode() + b"\n")
+            # ensure SSH key has correct permissions
+            os.chmod(ssh_key_path, 0o600)
+            ssh_command.extend(["-i", ssh_key_path])
+
+        self.env["GIT_SSH_COMMAND"] = " ".join(ssh_command)
 
     def clone_repository(self):
         """
@@ -25,62 +83,24 @@ def clone_repository(self):
         Returns:
             Path to the main script file to execute
         """
-        repo_url = self.git_cfg.url
-        if not repo_url:
-            raise UserException("Git repository URL is required")
 
         branch = self.git_cfg.branch or "main"
 
-        logging.info("Cloning git repository: %s", repo_url)
+        logging.info("Cloning git repository: %s", self.git_cfg.url)
 
         try:
             clone_args = ["git", "clone"]
 
             if branch:
                 clone_args.extend(["--branch", branch])
 
-            if self.git_cfg.ssh_keys.keys.encrypted_private and self.git_cfg.encrypted_token:
-                self.git_cfg.encrypted_token = None
-
-            if self.git_cfg.encrypted_token and repo_url.startswith("https://"):
-                repo_url = repo_url.replace("https://", f"https://x-token-auth:{self.git_cfg.encrypted_token}@")
-
-            clone_args.extend([repo_url, self.REPO_PATH])
-
-            env = os.environ.copy()
-
-            ssh_command = [
-                "ssh",
-                # the following lines could be used to disable strict host key checking, but it is better
-                # for security reasons to use the known_hosts file prepared in Dockerfile
-                # "-o",
-                # "StrictHostKeyChecking=no",
-                "-o",
-                "BatchMode=yes",  # do not ask for credentials when SSH auth fails
-                "-o",
-                "ConnectTimeout=30",
-                "-o",
-                "ServerAliveInterval=60",
-            ]
-
-            if self.git_cfg.ssh_keys.keys.encrypted_private:
-                ssh_key_path = os.path.expanduser("~/.ssh/github_private_key")
-                with open(ssh_key_path, "wb") as f:
-                    for line in self.git_cfg.ssh_keys.keys.encrypted_private.splitlines():
-                        f.write(line.encode() + b"\n")
-                # ensure SSH key has correct permissions
-                os.chmod(ssh_key_path, 0o600)
-                ssh_command.extend(["-i", ssh_key_path])
-            elif repo_url.startswith("git@") or repo_url.startswith("ssh://"):
-                logging.warning("SSH URL detected but no ssh_key_path provided. Trying default SSH configuration.")
-
-            env["GIT_SSH_COMMAND"] = " ".join(ssh_command)
+            clone_args.extend([self.repo_auth_url or self.git_cfg.url, self.REPO_PATH])
 
             process = subprocess.Popen(
                 clone_args,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
-                env=env,
+                env=self.env,
             )
             _, stderr = process.communicate()
 
@@ -110,33 +130,23 @@ def get_repository_branches(self):
             List of branch names
         """
         try:
-            repo_url = self.git_cfg.url
-            if not repo_url:
-                raise UserException("Git repository URL is required")
-
             branches_args = ["git", "ls-remote", "--heads"]
 
-            if self.git_cfg.ssh_keys.keys.encrypted_private and self.git_cfg.encrypted_token:
-                self.git_cfg.encrypted_token = None
-
-            if self.git_cfg.encrypted_token and repo_url.startswith("https://"):
-                repo_url = repo_url.replace("https://", f"https://x-token-auth:{self.git_cfg.encrypted_token}@")
-
-            branches_args.append(repo_url)
+            branches_args.append(self.repo_auth_url or self.git_cfg.url)
 
             process = subprocess.Popen(
                 branches_args,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
-                cwd=self.REPO_PATH,
+                env=self.env,
             )
             stdout, stderr = process.communicate()
 
             if process.returncode != 0:
                 raise UserException(f"Failed to get branches: {stderr.decode()}")
 
-            branches = [line.strip().split("refs/heads")[-1] for line in stdout.decode().splitlines() if line.strip()]
-            return branches
+            branches = [line.strip().split("refs/heads/")[-1] for line in stdout.decode().splitlines() if line.strip()]
+            return [{"value": b, "label": b} for b in branches]
 
         except Exception as e:
             raise UserException(f"Error getting repository branches: {str(e)}") from e