Add avro headers

Author: Fireye04

src/sasquatchbackpack/sasquatch.py

@@ -24,6 +24,7 @@
 from safir.kafka import (
     KafkaConnectionSettings,
     PydanticSchemaManager,
+    SchemaInfo,
     SchemaManagerSettings,
 )
 
@@ -179,14 +180,17 @@ async def _dispatch(
     except Exception as e:
         return e
 
-    await schema_manager.register_model(type(schema))
+    info: SchemaInfo = await schema_manager.register_model(type(schema))
 
     for record in records:
         avro: bytes = await schema_manager.serialize(
-            source.assemble_schema(record, namespace)
+            source.assemble_schema(record["value"], namespace)
         )
-
-        await publisher.publish(avro)
+        headers = {
+            "content-type": "avro",
+            "schema-id": str(info.schema_id),
+        }
+        await publisher.publish(avro, headers=headers)
     return None
 
 
@@ -229,7 +233,7 @@ def __init__(
             self.broker = broker_in if broker_in is not None else local_broker
         except NameError:
             if broker_in is None:
-                raise ValueError from None
+                raise ValueError("Kafka environment variables unset") from None
             self.broker = broker_in
 
         try:
@@ -239,7 +243,7 @@ def __init__(
             )
         except NameError:
             if manager_in is None:
-                raise ValueError from None
+                raise ValueError("Schema environment variable unset") from None
             self.manager = manager_in
 
     def create_topic(self) -> str:

tests/dispatcher_test.py

@@ -32,7 +32,7 @@ def __init__(self, current_records: list[dict[str, str]]) -> None:
 
     def assemble_schema(self, record: dict, namespace: str) -> AvroBaseModel:
         schema = {
-            "id": record["value"]["id"],
+            "id": record["id"],
             "namespace": namespace,
         }
         return TestSchema.parse_obj(data=schema)