connection init

Author: Fireye04

src/sasquatchbackpack/sasquatch.py

@@ -3,13 +3,16 @@
 __all__ = ["BackpackDispatcher", "DataSource", "DispatcherConfig"]
 
 import asyncio
+import json
 import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from string import Template
 
 import redis.asyncio as redis
 import requests
+from faststream.kafka import KafkaBroker
+from safir.kafka import KafkaConnectionSettings, SecurityProtocol
 
 # Code yoinked from https://github.com/lsst-sqre/
 # sasquatch/blob/main/examples/RestProxyAPIExample.ipynb
@@ -215,6 +218,25 @@ def _get_source_records(self) -> list[dict] | None:
             if self.redis.get(self.source.get_redis_key(record)) is None
         ]
 
+    async def direct_connect(self) -> None:
+        """Assemble a schema and payload from the given source,
+        and route data directly to kafka.
+        """
+        config = KafkaConnectionSettings(
+            bootstrap_servers=os.getenv("KAFKA_BOOTSTRAP_SERVERS", ""),
+            security_protocol=SecurityProtocol.SSL,
+        )
+        kafka_broker = KafkaBroker(**config.to_faststream_params())
+
+        records = self._get_source_records()
+        if records is None:
+            return
+        if len(records) == 0:
+            return
+
+        payload = json.dumps({"value_schema": self.schema, "records": records})
+        await kafka_broker.publish(payload, self.source.topic_name)
+
     def post(self) -> tuple[str, list]:
         """Assemble schema and payload from the given source, then
         makes a POST request to kafka.

src/sasquatchbackpack/sources/usgs/commands.py

@@ -1,6 +1,6 @@
 """USGS CLI."""
 
-import os
+import asyncio
 from datetime import UTC, datetime, timedelta
 
 import click
@@ -199,12 +199,11 @@ def usgs_earthquake_data(
 
     click.echo("Post mode enabled: Sending data...")
     click.echo(f"Querying redis at {backpack_dispatcher.redis.address}")
-    click.echo(
-        os.getenv(
-            "SASQUATCH_REST_PROXY_URL",
-            "no env var :(",
-        )
-    )
+    loop = asyncio.new_event_loop()
+    loop.run_until_complete(backpack_dispatcher.direct_connect())
+
+    click.echo("complete")
+
     result, records = backpack_dispatcher.post()
 
     if "Error" in result: