Clan_lib: add filtering by tag to list API

pkgs/clan-cli/clan_cli/machines/list.py

@@ -3,7 +3,6 @@ import logging
 
 from clan_lib.flake import Flake
 from clan_lib.machines.actions import list_machines
-from clan_lib.machines.list import query_machines_by_tags
 
 from clan_cli.completions import add_dynamic_completer, complete_tags
 
@@ -13,11 +12,7 @@ log = logging.getLogger(__name__)
 def list_command(args: argparse.Namespace) -> None:
     flake: Flake = args.flake
 
-    if args.tags:
-        for name in query_machines_by_tags(flake, args.tags):
-            print(name)
-    else:
-        for name in list_machines(flake):
+    for name in list_machines(flake, opts={"filter": {"tags": args.tags}}):
         print(name)
 
 

pkgs/clan-cli/clan_cli/machines/update.py

@@ -4,7 +4,7 @@ import sys
 
 from clan_lib.async_run import AsyncContext, AsyncOpts, AsyncRuntime
 from clan_lib.errors import ClanError
-from clan_lib.machines.list import list_full_machines
+from clan_lib.machines.list import list_full_machines, query_machines_by_tags
 from clan_lib.machines.machines import Machine
 from clan_lib.machines.suggestions import validate_machine_names
 from clan_lib.machines.update import deploy_machine
@@ -16,7 +16,6 @@ from clan_cli.completions import (
     complete_machines,
     complete_tags,
 )
-from clan_cli.machines.list import query_machines_by_tags
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_lib/machines/actions.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import TypedDict
 
 from clan_lib.api import API
 from clan_lib.errors import ClanError
@@ -10,15 +11,44 @@ from clan_lib.persist.inventory_store import InventoryStore
 from clan_lib.persist.util import set_value_by_path
 
 
+class MachineFilter(TypedDict):
+    tags: list[str]
+
+
+class ListOptions(TypedDict):
+    filter: MachineFilter
+
+
 @API.register
-def list_machines(flake: Flake) -> dict[str, InventoryMachine]:
+def list_machines(
+    flake: Flake, opts: ListOptions | None = None
+) -> dict[str, InventoryMachine]:
     """
-    List machines in the inventory for the UI.
+    List machines of a clan
+
+    Usage Example:
+
+    machines = list_machines(flake, {"filter": {"tags": ["foo" "bar"]}})
+
+    lists only machines that include both "foo" AND "bar"
+
     """
     inventory_store = InventoryStore(flake=flake)
     inventory = inventory_store.read()
 
     machines = inventory.get("machines", {})
+
+    if opts and opts.get("filter"):
+        filtered_machines = {}
+        filter_tags = opts.get("filter", {}).get("tags", [])
+
+        for machine_name, machine in machines.items():
+            machine_tags = machine.get("tags", [])
+            if all(ft in machine_tags for ft in filter_tags):
+                filtered_machines[machine_name] = machine
+
+        return filtered_machines
+
     return machines
 
 

pkgs/clan-cli/clan_lib/machines/list.py

@@ -31,21 +31,21 @@ def list_full_machines(flake: Flake) -> dict[str, Machine]:
     return res
 
 
-# TODO: Add filter to list_machines -> list_machines(flake, filter={tags=...})
-def query_machines_by_tags(flake: Flake, tags: list[str]) -> dict[str, Machine]:
+def query_machines_by_tags(
+    flake: Flake, tags: list[str]
+) -> dict[str, InventoryMachine]:
     """
     Query machines by their respective tags, if multiple tags are specified
     then only machines that have those respective tags specified will be listed.
     It is an intersection of the tags and machines.
     """
-    machines = list_full_machines(flake)
+    machines = list_machines(flake)
 
     filtered_machines = {}
-    for machine in machines.values():
-        inv_machine = get_machine(machine.flake, machine.name)
-        machine_tags = inv_machine.get("tags", [])
+    for machine_name, machine in machines.items():
+        machine_tags = machine.get("tags", [])
         if all(tag in machine_tags for tag in tags):
-            filtered_machines[machine.name] = machine
+            filtered_machines[machine_name] = machine
 
     return filtered_machines