Merge pull request #25 from Stackmasters/inventory-commands

Implement inventory commands

Author: parath

cycleops/__main__.py

@@ -3,6 +3,9 @@
 import typer
 
 from .client import cycleops_client
+from .environments import app as environments_app
+from .hostgroups import app as hostgroups_app
+from .hosts import app as hosts_app
 from .services import app as services_app
 from .setups import app as setups_app
 from .stacks import app as stacks_app
@@ -14,6 +17,11 @@
 cycleops.add_typer(stacks_app, name="stacks", help="Manage your stacks.")
 cycleops.add_typer(setups_app, name="setups", help="Manage your setups.")
 cycleops.add_typer(units_app, name="units", help="List all of the available units.")
+cycleops.add_typer(
+    environments_app, name="environments", help="Manage your environments."
+)
+cycleops.add_typer(hosts_app, name="hosts", help="Manage your hosts.")
+cycleops.add_typer(hostgroups_app, name="hostgroups", help="Manage your hostgroups.")
 
 
 @cycleops.callback()

cycleops/client.py

@@ -192,4 +192,75 @@ def delete(self, stack_id: int) -> Optional[Dict[str, Any]]:
         return self.client._request("DELETE", f"stacks/{stack_id}")
 
 
+class EnvironmentClient(SubClient):
+    """
+    Client for managing Cycleops environments.
+    """
+
+    def list(self) -> Optional[Dict[str, Any]]:
+        return self.client._request("GET", f"environments")
+
+
+class HostClient(SubClient):
+    """
+    Client for managing Cycleops hosts.
+    """
+
+    def list(self) -> Optional[Dict[str, Any]]:
+        return self.client._request("GET", f"hosts")
+
+    def retrieve(
+        self, host_id: Optional[int] = None, params: Optional[Dict[str, Any]] = None
+    ) -> Optional[Dict[str, Any]]:
+        if host_id:
+            return self.client._request("GET", f"hosts/{host_id}")
+
+        return self.client._request("GET", f"hosts", params=params)
+
+    def create(self, **kwargs: Any) -> Optional[Dict[str, Any]]:
+        payload: Dict[str, Any] = {k: v for (k, v) in kwargs.items() if v}
+
+        return self.client._request("POST", "hosts", payload)
+
+    def update(self, host_id: int, **kwargs: Any) -> Optional[Dict[str, Any]]:
+        payload: Dict[str, Any] = {k: v for (k, v) in kwargs.items() if v}
+
+        return self.client._request("PATCH", f"hosts/{host_id}", payload)
+
+    def delete(self, host_id: int) -> Optional[Dict[str, Any]]:
+        return self.client._request("DELETE", f"hosts/{host_id}")
+
+
+class HostgroupClient(SubClient):
+    """
+    Client for managing Cycleops hostgroups.
+    """
+
+    def list(self) -> Optional[Dict[str, Any]]:
+        return self.client._request("GET", f"hostgroups")
+
+    def retrieve(
+        self,
+        hostgroup_id: Optional[int] = None,
+        params: Optional[Dict[str, Any]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        if hostgroup_id:
+            return self.client._request("GET", f"hostgroups/{hostgroup_id}")
+
+        return self.client._request("GET", f"hostgroups", params=params)
+
+    def create(self, **kwargs: Any) -> Optional[Dict[str, Any]]:
+        payload: Dict[str, Any] = {k: v for (k, v) in kwargs.items() if v}
+
+        return self.client._request("POST", "hostgroups", payload)
+
+    def update(self, hostgroup_id: int, **kwargs: Any) -> Optional[Dict[str, Any]]:
+        payload: Dict[str, Any] = {k: v for (k, v) in kwargs.items() if v}
+
+        return self.client._request("PATCH", f"hostgroups/{hostgroup_id}", payload)
+
+    def delete(self, hostgroup_id: int) -> Optional[Dict[str, Any]]:
+        return self.client._request("DELETE", f"hostgroups/{hostgroup_id}")
+
+
 cycleops_client: Client = Client()

cycleops/environments.py

@@ -0,0 +1,36 @@
+import typer
+from rich import print
+from rich.table import Table
+
+from .client import EnvironmentClient, cycleops_client
+from .exceptions import NotFound
+from .utils import display_error_message
+
+app = typer.Typer()
+
+environment_client: EnvironmentClient = EnvironmentClient(cycleops_client)
+
+
+@app.command()
+def list() -> None:
+    """
+    List all of the available environments.
+    """
+
+    try:
+        environments = environment_client.list()
+
+        if not environments:
+            raise NotFound("No environments available")
+
+        table = Table(show_header=True, leading=True)
+        table.add_column("ID", width=5)
+        table.add_column("Name", width=30)
+
+        for environment in environments:
+            table.add_row(str(environment["id"]), environment["name"])
+
+        print(table)
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Exit(code=1)

cycleops/hostgroups.py

@@ -0,0 +1,152 @@
+from typing import Any, Dict, List, Optional
+
+import typer
+from rich import print
+
+from .client import HostgroupClient, cycleops_client
+from .exceptions import NotFound
+from .utils import display_error_message, display_success_message
+
+app = typer.Typer()
+
+hostgroup_client: HostgroupClient = HostgroupClient(cycleops_client)
+
+
+@app.command()
+def list() -> None:
+    """
+    List all of the available hostgroups.
+    """
+
+    try:
+        hostgroups = hostgroup_client.list()
+
+        if not hostgroups:
+            raise NotFound("No hostgroups available")
+
+        print(hostgroups)
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Exit(code=1)
+
+
+@app.command()
+def retrieve(
+    hostgroup_identifier: str = typer.Argument(
+        ..., help="The ID or name of the hostgroup. Names take precedence."
+    ),
+) -> None:
+    """
+    Retrieve the hostgroup specified by the given ID.
+    """
+
+    try:
+        hostgroup = get_hostgroup(hostgroup_identifier)
+
+        print(hostgroup)
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Exit(code=1)
+
+
+@app.command()
+def create(
+    name: str = typer.Option(
+        ...,
+        help="The name of the hostgroup.",
+    ),
+    environment_id: int = typer.Option(None, help="The ID of the environment."),
+    hosts: Optional[List[int]] = typer.Option(
+        None,
+        "--host-id",
+        help="The ID of the host.",
+    ),
+) -> None:
+    """
+    Create a hostgroup with the specified option values.
+    """
+
+    try:
+        hostgroup_client.create(
+            name=name,
+            environment=environment_id,
+            hosts=hosts,
+        )
+
+        display_success_message(f"Hostgroup {name} has been created")
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Abort()
+
+
+@app.command()
+def update(
+    hostgroup_identifier: str = typer.Argument(
+        ..., help="The ID or name of the hostgroup. Names take precedence."
+    ),
+    name: Optional[str] = typer.Option(
+        None,
+        help="The name of the hostgroup.",
+    ),
+    environment_id: Optional[int] = typer.Option(
+        None, help="The ID of the environment."
+    ),
+    hosts: Optional[List[int]] = typer.Option(
+        None,
+        "--host-id",
+        help="The ID of the host.",
+    ),
+) -> None:
+    """
+    Update a hostgroup with the specified option values.
+    """
+
+    try:
+        hostgroup = get_hostgroup(hostgroup_identifier)
+
+        hostgroup_client.update(
+            hostgroup["id"],
+            name=name,
+            environment=environment_id,
+            hosts=hosts,
+        )
+
+        display_success_message(f"Hostgroup {hostgroup_identifier} has been updated")
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Abort()
+
+
+@app.command()
+def delete(
+    hostgroup_identifier: str = typer.Argument(
+        ..., help="The ID or name of the hostgroup. Names take precedence."
+    ),
+) -> None:
+    """
+    Delete the hostgroup specified by the given name.
+    """
+
+    try:
+        hostgroup = get_hostgroup(hostgroup_identifier)
+
+        hostgroup_client.delete(hostgroup["id"])
+        display_success_message(f"Hostgroup {hostgroup_identifier} has been deleted")
+    except Exception as error:
+        display_error_message(error)
+        raise typer.Abort()
+
+
+def get_hostgroup(hostgroup_identifier: str) -> Optional[Dict[str, Any]]:
+    """
+    Retrieves a hostgroup with either a name or ID. Names take precedence.
+    """
+
+    hostgroup = hostgroup_client.retrieve(params={"name": hostgroup_identifier})
+
+    if len(hostgroup) == 1:
+        return hostgroup[0]
+
+    hostgroup = hostgroup_client.retrieve(hostgroup_identifier)
+
+    return hostgroup