Deployment refactor

src/sasquatchbackpack/cli.py

@@ -3,6 +3,7 @@
 import click
 
 from sasquatchbackpack import sasquatch, sources
+from sasquatchbackpack.scripts import usgs
 
 DEFAULT_RADIUS = 400
 
@@ -177,12 +178,13 @@ def usgs_earthquake_data(
     """
     total_duration = timedelta(duration[0], 0, 0, 0, 0, duration[1], 0)
 
-    source = sources.usgs_source(
-        total_duration, radius, coords, magnitude_bounds
+    results = usgs.search_api(
+        total_duration,
+        radius,
+        coords,
+        magnitude_bounds,
     )
 
-    results = source.get_results()
-
     if len(results) > 0:
         click.secho("SUCCESS!", fg="green")
         click.echo("------")
@@ -199,9 +201,16 @@ def usgs_earthquake_data(
     if not test:
         click.echo("Sending data...")
 
-        poster = sasquatch.sasquatch_deploy(source)
+        config = sources.USGSConfig(
+            total_duration, radius, coords, magnitude_bounds
+        )
+        source = sources.USGSSource(config)
+
+        poster = sasquatch.BackpackDispatcher(
+            source, sasquatch.DispatcherConfig()
+        )
 
-        click.echo(poster.send_data())
+        click.echo(poster.post())
 
 
 if __name__ == "__main__":

src/sasquatchbackpack/sasquatch.py

@@ -1,4 +1,5 @@
-from string import Template
+import json
+from dataclasses import dataclass
 
 import requests
 
@@ -8,35 +9,61 @@
 # sasquatch/blob/main/examples/RestProxyAPIExample.ipynb
 
 
-class sasquatch_deploy:
+@dataclass
+class DispatcherConfig:
+    """Class containing relevant configuration information for the
+    BackpackDispatcher
+    """
+
+    sasquatch_rest_proxy_url = (
+        "https://data-int.lsst.cloud/sasquatch-rest-proxy"
+    )
+
+
+class BackpackDispatcher:
     """A class to send backpack data to kafka.
 
     Parameters
     ----------
     source
-        data_source containing the payload and other important,
-        individulized data
+        DataSource containing schema and record data to be
+        posted to remote
+    config
+        DispatcherConfig to transmit other relevant information to
+        the Dispatcher
     """
 
-    def __init__(self, source: sources.data_source):
+    def __init__(self, source: sources.DataSource, config: DispatcherConfig):
         self.source = source
+        self.config = config
+        self.schema = source.load_schema()
+        self.namespace = self.get_namespace()
 
-    def create_kafka_topic(self) -> str:
+    def get_namespace(self) -> str:
+        """Sorts the schema and returns the namespace value
+
+        Returns
+        -------
+        namespace
+            provided namespace from the schema file
+        """
+        json_schema = json.loads(self.schema)
+
+        return json_schema["namespace"]
+
+    def create_topic(self) -> str:
         """Creates kafka topic based off data from provided source
 
         Returns
         -------
         response text
             The results of the POST request in string format
         """
-        sasquatch_rest_proxy_url = (
-            "https://data-int.lsst.cloud/sasquatch-rest-proxy"
-        )
-
         headers = {"content-type": "application/json"}
 
         r = requests.get(
-            f"{sasquatch_rest_proxy_url}/v3/clusters", headers=headers
+            f"{self.config.sasquatch_rest_proxy_url}/v3/clusters",
+            headers=headers,
         )
 
         cluster_id = r.json()["data"][0]["cluster_id"]
@@ -45,39 +72,30 @@ def create_kafka_topic(self) -> str:
         # factor of 3 by default, this configuration is fixed for the
         # Sasquatch Kafka cluster.
         topic_config = {
-            "topic_name": f"{self.source.namespace}."
-            + f"{self.source.topic_name}",
+            "topic_name": f"{self.namespace}." + f"{self.source.topic_name}",
             "partitions_count": 1,
             "replication_factor": 3,
         }
 
         headers = {"content-type": "application/json"}
 
         response = requests.post(
-            f"{sasquatch_rest_proxy_url}/v3/clusters/{cluster_id}/topics",
+            f"{self.config.sasquatch_rest_proxy_url}/v3/clusters/"
+            + f"{cluster_id}/topics",
             json=topic_config,
             headers=headers,
         )
         return response.text
 
-    def send_data(self) -> dict:
-        """Assemble schema and payload, then make a POST request to kafka
+    def post(self) -> dict:
+        """Assemble schema and payload from the given source, then
+        makes a POST request to kafka
 
         Returns
         -------
         response text
             The results of the POST request in string format
         """
-        with open(self.source.schema_directory, "r") as file:
-            template = Template(file.read())
-
-        value_schema = template.substitute(
-            {
-                "namespace": self.source.namespace,
-                "topic_name": self.source.topic_name,
-            }
-        )
-
         # Currently unused, TODO: Uncomment when POSTing begins
 
         # url = f"{sasquatch_rest_proxy_url}/topics/"
@@ -92,7 +110,7 @@ def send_data(self) -> dict:
 
         records = self.source.get_records()
 
-        payload = {"value_schema": value_schema, "records": records}
+        payload = {"value_schema": self.schema, "records": records}
 
         # Temporarily returns payload instead of making full
         # POST request.

src/sasquatchbackpack/schemas/usgs.avsc

@@ -1 +1 @@
-{"namespace": "$namespace", "type": "record", "name": "$topic_name", "description": "Collection of earthquakes near the summit", "fields": [{"name": "timestamp", "type": "long"}, {"name": "id", "type": "str", "description": "unique earthquake id"}, {"name": "latitude",  "type": "float", "units": "Degrees"}, {"name": "longitude", "type": "float", "units": "Degrees"}, {"name": "depth", "type": "float", "units": "Km"}, {"name": "magnitude", "type": "float", "units": "Richter Magnitudes"}]}
+{"namespace": "lsst.example", "type": "record", "name": "$topic_name", "description": "Collection of earthquakes near the summit", "fields": [{"name": "timestamp", "type": "long"}, {"name": "id", "type": "str", "description": "unique earthquake id"}, {"name": "latitude",  "type": "float", "units": "Degrees"}, {"name": "longitude", "type": "float", "units": "Degrees"}, {"name": "depth", "type": "float", "units": "Km"}, {"name": "magnitude", "type": "float", "units": "Richter Magnitudes"}]}