added --regex option to filter hosts aliases based on regex pattern

Author: Blackmamoth

.github/workflows/test.pypi.yml

@@ -0,0 +1,52 @@
+name: Publish to Test PyPI
+
+on:
+  push:
+    branches:
+      - dev
+
+jobs:
+  build_package:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+
+      - name: Build release distribution
+        run: |
+          uv build
+
+      - name: Upload distribution
+        uses: actions/upload-artifact@v4
+        with:
+          name: test-release-dists
+          path: dist/
+
+  test-pypi-publish:
+    runs-on: ubuntu-latest
+    needs: build_package
+    permissions:
+      id-token: write
+
+    environment:
+      name: test-pypi
+
+    steps:
+      - name: Retrieve release distributions
+        uses: actions/download-artifact@v4
+        with:
+          name: test-release-dists
+          path: dist/
+
+      - name: Publish release distributions to Test PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          packages-dir: dist/
+          repository-url: https://test.pypi.org/legacy/

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sshsync"
-version = "0.10.0"
+version = "0.11.0"
 description = "sshsync is a CLI tool to run shell commands across multiple servers via SSH, either on specific groups or all servers. It also supports pushing and pulling files to and from remote hosts."
 readme = "README.md"
 authors = [

src/sshsync/cli.py

@@ -10,6 +10,7 @@
     add_hosts_to_group,
     assign_groups_to_hosts,
     check_path_exists,
+    is_valid_regex,
     list_configuration,
     print_dry_run_results,
     print_error,
@@ -45,6 +46,9 @@ def all(
     try:
         config = Config()
 
+        if not config.hosts:
+            print_error("No hosts", True)
+
         ssh_client = SSHClient()
         if dry_run:
             dry_run_results = ssh_client.begin_dry_run_exec(
@@ -64,6 +68,9 @@ def all(
 def group(
     name: str = typer.Argument(..., help="Name of the host group to target."),
     cmd: str = typer.Argument(..., help="The shell command to execute on the group."),
+    regex: str = typer.Option(
+        "", help="Filter group members by matching alias with a regex pattern."
+    ),
     timeout: int = typer.Option(
         10, help="Timeout in seconds for SSH command execution."
     ),
@@ -77,12 +84,19 @@ def group(
     Args:
         name (str): The name of the host group to target.
         cmd (str): The shell command to execute remotely.
+        regex (str): Filter group members by matching alias with regex pattern.
         timeout (int): Timeout (in seconds) for both SSH connection and command execution.
         dry_run (bool): Show command and host info without executing.
     """
     try:
+        if regex and not is_valid_regex(regex):
+            print_error("Invalid regex", True)
+
         config = Config()
-        hosts = config.get_hosts_by_group(name)
+        hosts = config.get_hosts_by_group(name, regex)
+
+        if not hosts:
+            print_error("Invalid group", True)
 
         ssh_client = SSHClient()
         if dry_run:
@@ -163,6 +177,9 @@ def push(
     ),
     all: bool = typer.Option(False, help="Push to all configured hosts."),
     group: str = typer.Option("", help="Push to a specific group of hosts."),
+    regex: str = typer.Option(
+        "", help="Filter group members by matching alias with a regex pattern."
+    ),
     host: str = typer.Option("", help="Push to a single specific host."),
     recurse: bool = typer.Option(
         False, help="Recursively push a directory and its contents."
@@ -182,17 +199,33 @@ def push(
         remote_path (str): The destination path on the remote host(s).
         all (bool): Push to all hosts.
         group (str): Push to a specified group of hosts.
+        regex (str): Filter group members by matching alias with regex pattern.
         host (str): Push to a specified individual host.
         recurse (bool): If True, recursively push a directory and all its contents.
         dry_run (bool): Show transfer and host info without executing.
     """
-    options = [all, bool(group != ""), bool(host != "")]
+    has_all = all
+    has_group = bool(group != "")
+    has_host = bool(host != "")
+    has_regex = bool(regex != "")
+
+    if has_regex and not is_valid_regex(regex):
+        print_error("Invalid regex", True)
+
+    options = [has_all, has_group, has_host]
+
     if sum(options) != 1:
         print_error(
             "You must specify exactly one of --all, --group, or --host.",
             True,
         )
 
+    if has_regex and not has_group:
+        print_error(
+            "--regex can only be used with --group.",
+            True,
+        )
+
     if not check_path_exists(local_path):
         print_error(f"Path ({local_path}) does not exist", True)
 
@@ -205,7 +238,7 @@ def push(
             config.configured_hosts()
             if all
             else (
-                config.get_hosts_by_group(group)
+                config.get_hosts_by_group(group, regex)
                 if group
                 else [host_obj]
                 if host_obj is not None
@@ -241,6 +274,9 @@ def pull(
     ),
     all: bool = typer.Option(False, "--all", help="Pull from all configured hosts."),
     group: str = typer.Option("", help="Pull from a specific group of hosts."),
+    regex: str = typer.Option(
+        "", help="Filter group members by matching alias with a regex pattern."
+    ),
     host: str = typer.Option("", help="Pull from a single specific host."),
     recurse: bool = typer.Option(
         False, help="Recursively pull a directory and its contents."
@@ -260,17 +296,33 @@ def pull(
         local_path (str): The local file path.
         all (bool): Pull from all hosts.
         group (str): Pull from a specified group of hosts.
+        regex (str): Filter group members by matching alias with regex pattern.
         host (str): Pull from a specified individual host.
         recurse (bool): If True, recursively pull directories and all their contents.
         dry_run (bool): Show transfer and host info without executing.
     """
-    options = [all, bool(group), bool(host)]
+    has_all = all
+    has_group = bool(group != "")
+    has_host = bool(host != "")
+    has_regex = bool(regex != "")
+
+    if has_regex and not is_valid_regex(regex):
+        print_error("Invalid regex", True)
+
+    options = [has_all, has_group, has_host]
+
     if sum(options) != 1:
         print_error(
             "You must specify exactly one of --all, --group, or --host.",
             True,
         )
 
+    if has_regex and not has_group:
+        print_error(
+            "--regex can only be used with --group.",
+            True,
+        )
+
     if not check_path_exists(local_path):
         print_error(f"Path ({local_path}) does not exist", True)
 
@@ -283,7 +335,7 @@ def pull(
             config.configured_hosts()
             if all
             else (
-                config.get_hosts_by_group(group)
+                config.get_hosts_by_group(group, regex)
                 if group
                 else [host_obj]
                 if host_obj is not None

src/sshsync/config.py

@@ -1,3 +1,4 @@
+import re
 from pathlib import Path
 
 import structlog
@@ -154,19 +155,22 @@ def _save_yaml(self) -> None:
                 indent=4,
             )
 
-    def get_hosts_by_group(self, group: str) -> list[Host]:
+    def get_hosts_by_group(self, group: str, regex: str = "") -> list[Host]:
         """Return all hosts that belong to the specified group.
 
         Args:
             group (str): Group name to filter hosts by.
+            regex (str): Only include host aliases with the matching regex
 
         Returns:
             list[Host]: Hosts that are members of the group.
         """
         return [
             host
             for host in self.hosts
-            if group in host.groups and host.alias != "default"
+            if group in host.groups
+            and host.alias != "default"
+            and (not regex or re.search(regex, host.alias))
         ]
 
     def get_host_by_name(self, name: str) -> Host | None:

src/sshsync/utils.py

@@ -1,5 +1,6 @@
 import asyncio
 import ipaddress
+import re
 import socket
 from pathlib import Path
 
@@ -75,6 +76,15 @@ def is_valid_ip(ip: str) -> bool:
         return False
 
 
+def is_valid_regex(pattern: str) -> bool:
+    "Check if the string is a valid regex pattern"
+    try:
+        re.compile(pattern)
+        return True
+    except Exception:
+        return False
+
+
 def get_host_name_or_ip() -> str:
     """Prompt the user to enter a valid hostname or ip address"""
     while True: