Add publisher

Author: Fireye04

src/sasquatchbackpack/sasquatch.py

@@ -223,6 +223,7 @@ async def direct_connect(self) -> None:
         """
         config = KafkaConnectionSettings()
         kafka_broker = KafkaBroker(**config.to_faststream_params())
+        prepared_publisher = kafka_broker.publisher(self.source.topic_name)
 
         records = self._get_source_records()
         if records is None:
@@ -231,7 +232,13 @@ async def direct_connect(self) -> None:
             return
 
         payload = json.dumps({"value_schema": self.schema, "records": records})
-        await kafka_broker.publish(payload, self.source.topic_name)
+        await prepared_publisher.publish(
+            payload,
+            headers={
+                "Content-Type": "application/vnd.kafka.avro.v2+json",
+                "Accept": "application/vnd.kafka.v2+json",
+            },
+        )
 
     def post(self) -> tuple[str, list]:
         """Assemble schema and payload from the given source, then

src/sasquatchbackpack/sources/usgs/commands.py

@@ -4,7 +4,6 @@
 from datetime import UTC, datetime, timedelta
 
 import click
-import nest_asyncio
 
 from sasquatchbackpack import sasquatch
 from sasquatchbackpack.sources.usgs import scripts
@@ -201,7 +200,6 @@ def usgs_earthquake_data(
     click.echo("Post mode enabled: Sending data...")
     click.echo(f"Querying redis at {backpack_dispatcher.redis.address}")
 
-    nest_asyncio.apply()
     loop = asyncio.new_event_loop()
     loop.run_until_complete(backpack_dispatcher.direct_connect())