Add subcommands for consumption types to cli

piket_client/cli.py

@@ -1,5 +1,11 @@
 import click
-from piket_client.model import ServerStatus, NetworkError, Consumption, Settlement
+from piket_client.model import (
+    ServerStatus,
+    NetworkError,
+    Consumption,
+    Settlement,
+    ConsumptionType,
+)
 from prettytable import PrettyTable
 
 
@@ -99,6 +105,73 @@ def output_settlement_info(s: Settlement) -> None:
     print(table)
 
 
+@cli.group()
+def consumption_types():
+    pass
+
+
+@consumption_types.command("list")
+def list_consumption_types() -> None:
+    active = ConsumptionType.get_all(active=True)
+    inactive = ConsumptionType.get_all(active=False)
+
+    if isinstance(active, NetworkError) or isinstance(inactive, NetworkError):
+        print_error("Could not get consumption types!")
+        return
+
+    table = PrettyTable()
+    table.field_names = ["ID", "Name", "Active"]
+    table.sortby = "ID"
+
+    for ct in active + inactive:
+        table.add_row([ct.consumption_type_id, ct.name, ct.active])
+
+    print(table)
+
+
+@consumption_types.command("create")
+@click.argument("name")
+def create_consumption_type(name: str) -> None:
+    ct = ConsumptionType(name=name).create()
+
+    if not isinstance(ct, NetworkError):
+        print_ok(f'Created consumption type "{name}" with ID {ct.consumption_type_id}.')
+
+
+@consumption_types.command("activate")
+@click.argument("consumption_type_id", type=click.INT)
+def activate_consumption_type(consumption_type_id: int) -> None:
+    ct = ConsumptionType.get(consumption_type_id)
+
+    if isinstance(ct, NetworkError):
+        print_error(f"Could not get ConsumptionType: {ct.value}")
+        return
+
+    result = ct.set_active(True)
+
+    if not isinstance(result, NetworkError):
+        print_ok(
+            f"Consumption type {ct.consumption_type_id} ({ct.name}) is now active."
+        )
+
+
+@consumption_types.command("deactivate")
+@click.argument("consumption_type_id", type=click.INT)
+def deactivate_consumption_type(consumption_type_id: int) -> None:
+    ct = ConsumptionType.get(consumption_type_id)
+
+    if isinstance(ct, NetworkError):
+        print_error(f"Could not get ConsumptionType: {ct.value}")
+        return
+
+    result = ct.set_active(False)
+
+    if not isinstance(result, NetworkError):
+        print_ok(
+            f"Consumption type {ct.consumption_type_id} ({ct.name}) is now inactive."
+        )
+
+
 def print_ok(msg: str) -> None:
     click.echo(click.style(msg, fg="green"))
 

piket_client/model.py

@@ -327,85 +327,84 @@ class ConsumptionType(NamedTuple):
     name: str
     consumption_type_id: Optional[int] = None
     icon: Optional[str] = None
+    active: bool = True
 
-    def create(self) -> Optional[ConsumptionType]:
+    def create(self) -> Union[ConsumptionType, NetworkError]:
         """ Create a new ConsumptionType from the current attributes. As tuples
         are immutable, a new ConsumptionType with the correct id is returned.
         """
-        req = requests.post(
-            urljoin(SERVER_URL, "consumption_types"),
-            json={"consumption_type": {"name": self.name, "icon": self.icon}},
-        )
-
         try:
-            data = req.json()
-        except ValueError:
-            LOG.error(
-                "Did not get JSON on adding ConsumptionType (%s): %s",
-                req.status_code,
-                req.content,
+            req = requests.post(
+                urljoin(SERVER_URL, "consumption_types"),
+                json={"consumption_type": {"name": self.name, "icon": self.icon}},
             )
-            return None
 
-        if "error" in data or req.status_code != 201:
-            LOG.error(
-                "Could not create ConsumptionType (%s): %s", req.status_code, data
-            )
-            return None
+            req.raise_for_status()
+            data = req.json()
+            return ConsumptionType.from_dict(data["consumption_type"])
+
+        except requests.ConnectionError as e:
+            LOG.exception(e)
+            return NetworkError.ConnectionFailure
+
+        except requests.HTTPError as e:
+            LOG.exception(e)
+            return NetworkError.HttpFailure
+
+        except ValueError as e:
+            LOG.exception(e)
+            return NetworkError.InvalidData
 
-        return ConsumptionType.from_dict(data["consumption_type"])
 
     @classmethod
-    def get(cls, consumption_type_id: int) -> Optional[ConsumptionType]:
+    def get(cls, consumption_type_id: int) -> Union[ConsumptionType, NetworkError]:
         """ Retrieve a ConsumptionType by id. """
-        req = requests.get(
-            urljoin(SERVER_URL, f"/consumption_types/{consumption_type_id}")
-        )
-
         try:
+            req = requests.get(
+                urljoin(SERVER_URL, f"/consumption_types/{consumption_type_id}")
+            )
+            req.raise_for_status()
             data = req.json()
 
-            if "error" in data:
-                LOG.warning(
-                    "Could not get consumption type %s (%s): %s",
-                    consumption_type_id,
-                    req.status_code,
-                    data,
-                )
-                return None
+        except requests.ConnectionError as e:
+            LOG.exception(e)
+            return NetworkError.ConnectionFailure
 
-            return cls.from_dict(data["consumption_type"])
+        except requests.HTTPError as e:
+            LOG.exception(e)
+            return NetworkError.HttpFailure
 
-        except ValueError:
-            LOG.error(
-                "Did not get JSON from server on getting consumption type (%s): %s",
-                req.status_code,
-                req.content,
-            )
-            return None
+        except ValueError as e:
+            LOG.exception(e)
+            return NetworkError.InvalidData
 
-    @classmethod
-    def get_all(cls) -> Optional[List[ConsumptionType]]:
-        """ Get all active ConsumptionTypes. """
-        req = requests.get(urljoin(SERVER_URL, "/consumption_types"))
+        return cls.from_dict(data["consumption_type"])
 
+    @classmethod
+    def get_all(cls, active: bool = True) -> Union[List[ConsumptionType], NetworkError]:
+        """ Get the list of ConsumptionTypes. """
         try:
+            req = requests.get(
+                urljoin(SERVER_URL, "/consumption_types"),
+                params={"active": int(active)},
+            )
+            req.raise_for_status()
+
             data = req.json()
 
-            if "error" in data:
-                LOG.warning(
-                    "Could not get consumption types (%s): %s", req.status_code, data
-                )
+        except requests.ConnectionError as e:
+            LOG.exception(e)
+            return NetworkError.ConnectionFailure
 
-            return [cls.from_dict(item) for item in data["consumption_types"]]
+        except requests.HTTPError as e:
+            LOG.exception(e)
+            return NetworkError.HttpFailure
 
-        except ValueError:
-            LOG.error(
-                "Did not get JSON from server on getting ConsumptionTypes (%s): %s",
-                req.status_code,
-                req.content,
-            )
-            return None
+        except ValueError as e:
+            LOG.exception(e)
+            return NetworkError.InvalidData
+
+        return [cls.from_dict(x) for x in data["consumption_types"]]
 
     @classmethod
     def from_dict(cls, data: dict) -> "ConsumptionType":
@@ -414,8 +413,33 @@ class ConsumptionType(NamedTuple):
             name=data["name"],
             consumption_type_id=data["consumption_type_id"],
             icon=data.get("icon"),
+            active=data["active"],
         )
 
+    def set_active(self, active: bool) -> Union[ConsumptionType, NetworkError]:
+        """Update the 'active' attribute."""
+        try:
+            req = requests.patch(
+                urljoin(SERVER_URL, f"/consumption_types/{self.consumption_type_id}"),
+                json={"consumption_type": {"active": active}},
+            )
+            req.raise_for_status()
+            data = req.json()
+
+        except requests.ConnectionError as e:
+            LOG.exception(e)
+            return NetworkError.ConnectionFailure
+
+        except requests.HTTPError as e:
+            LOG.exception(e)
+            return NetworkError.HttpFailure
+
+        except ValueError as e:
+            LOG.exception(e)
+            return NetworkError.InvalidData
+
+        return self.from_dict(data["consumption_type"])
+
 
 class Consumption(NamedTuple):
     """ Represents a stored Consumption. """