Finished automatic topic creation

Author: Fireye04

src/sasquatchbackpack/sasquatch.py

@@ -185,9 +185,7 @@ def _get_topic_exists(self) -> bool | str:
 
         return bool(any(topic["topic_name"] == topic_name for topic in topics))
 
-    def _handle_topic_exists(
-        self, request: dict[str, dict[str, str]]
-    ) -> bool | None:
+    def _topic_exists(self, request: dict[str, dict[str, str]]) -> bool | None:
         """Handle errors and format responses for topic checking.
 
         Parameters
@@ -267,7 +265,10 @@ def _post_new_topic(self) -> dict[str, str]:
                 "message": f"Error Creating Topic during POST: {e}",
             }
 
-        return {"status": "Success", "message": response.text}
+        return {
+            "status": "Success",
+            "message": "A new kafka topic has been created.",
+        }
 
     def _get_source_records(self) -> list[dict] | None:
         """Get source records and check the redis server for any
@@ -299,6 +300,20 @@ def _get_source_records(self) -> list[dict] | None:
             if self.redis.get(self.source.get_redis_key(record)) is None
         ]
 
+    def _check_success(self, response: dict) -> None:
+        """Check whether the request has suceeded.
+
+        Parameters
+        ----------
+        response: dict
+            response object from post().
+        """
+        for item in response["requests"]:
+            if response["requests"][item]["status"] == "Error":
+                response["succeeded"] = False
+                break
+            response["succeeded"] = True
+
     def post(self, *, force_post: bool = False) -> str:
         """Assemble schema and payload from the given source, then
         makes a POST request to kafka.
@@ -337,19 +352,19 @@ def post(self, *, force_post: bool = False) -> str:
             },
             "records": [],
         }
-        requests = response["requests"]
+        reqs = response["requests"]
 
-        topic_exists = self._handle_topic_exists(requests)
+        topic_exists = self._topic_exists(reqs)
 
         if topic_exists is None:
             return json.dumps(response)
 
         if force_post or not topic_exists:
-            requests["create_topic"] = self._post_new_topic()
-            if requests["create_topic"]["status"] == "Error":
+            reqs["create_topic"] = self._post_new_topic()
+            if reqs["create_topic"]["status"] == "Error":
                 return json.dumps(response)
         else:
-            requests["create_topic"] = {
+            reqs["create_topic"] = {
                 "status": "Success",
                 "message": (
                     "Relevant topic already exists in kafka, "
@@ -360,19 +375,22 @@ def post(self, *, force_post: bool = False) -> str:
         records = self._get_source_records()
 
         if records is None:
-            requests["write_values"] = {
+            reqs["write_values"] = {
                 "status": "Warning",
                 "message": "No entries found, aborting POST request",
             }
             return json.dumps(response)
 
+        response["records"] = records
+
         if len(records) == 0:
-            requests["write_values"] = {
+            reqs["write_values"] = {
                 "status": "Warning",
                 "message": (
                     "All entries already present, aborting POST request"
                 ),
             }
+            self._check_success(response)
             return json.dumps(response)
 
         payload = {"value_schema": self.schema, "records": records}
@@ -411,6 +429,9 @@ def post(self, *, force_post: bool = False) -> str:
 
         response["write_values"] = {
             "status": "Success",
-            "message": post_response.text,
+            "message": "Data successfully sent!",
         }
+
+        self._check_success(response)
+
         return json.dumps(response)

src/sasquatchbackpack/sources/usgs/commands.py

@@ -1,5 +1,6 @@
 """USGS CLI."""
 
+import json
 from datetime import UTC, datetime, timedelta
 
 import click
@@ -13,9 +14,6 @@
 
 DEFAULT_MAGNITUDE_BOUNDS = (2, 10)
 
-# ruff: noqa:TD002
-# ruff: noqa:TD003
-
 
 def check_duration(
     ctx: click.Context, param: dict, value: tuple[int, int]
@@ -177,8 +175,6 @@ def usgs_earthquake_data(
         magnitude_bounds,
     )
 
-    # TODO: Add CLI feedback on what items are cached to redis
-    # and which are new
     if len(results) > 0:
         click.secho("SUCCESS!", fg="green")
         click.echo("------")
@@ -199,12 +195,56 @@ def usgs_earthquake_data(
     click.echo("Post mode enabled: Sending data...")
     click.echo(f"Querying redis at {backpack_dispatcher.redis.address}")
 
-    result, records = backpack_dispatcher.post()
+    response = json.loads(backpack_dispatcher.post())
+    _handle_post_response(response)
+
+
+def _handle_post_response(response: dict) -> None:
+    """Handle click response to post request output.
+
+    Parameters
+    ----------
+    response: dict
+        json loaded dictionary recieved from a backpack_dispatcher post.
+    """
+    if not response["succeeded"]:
+        failed: tuple[str, dict[str, str]] = ("none", {"none": "none"})
+        for name in response["requests"]:
+            item: dict = response["requests"][name]
+            click.echo(f"{name}: {item['message']} - Status: {item['status']}")
+            if item["status"] == "Error":
+                failed = (name, item)
+                break
+
+        if failed[0] == "none":
+            click.secho("Unknown error encountered", fg="red")
+            click.echo(json.dumps(response))
+            return
+
+        if failed[0] == "check_topic":
+            click.secho(
+                "Tip: You can use force_post=True to bypass topic checks.",
+                fg="yellow",
+            )
+        return
+
+    click.echo("------")
+    for name in response["requests"]:
+        item: dict = response["requests"][name]
+        if item["status"] == "Warning":
+            click.secho(
+                f"Warning! {name}: {item['message']}",
+                fg="yellow",
+            )
+            continue
+        click.secho(
+            f"{name}: {item['message']} - Status: {item['status']}", fg="green"
+        )
 
-    if "Error" in result:
-        click.secho(result, fg="red")
-    elif "Warning" in result:
-        click.secho(result, fg="yellow")
+    records: list = response["records"]
+    if len(records) == 0:
+        click.secho("Success!", fg="green")
+        click.echo("No data has been sent.")
     else:
         click.secho("Data successfully sent!", fg="green")
         click.echo("The following items were added to Kafka:")
@@ -223,9 +263,8 @@ def usgs_earthquake_data(
                 f"{value['depth']} km "
                 f"M{value['magnitude']}"
             )
-        click.echo("------")
+    click.echo("------")
 
-        click.echo(
-            "All entries missing from this list "
-            "have been identified as already present in Kafka."
-        )
+    click.echo(
+        "All missing entries have been identified as already present in Kafka."
+    )