Adds a try to run the validate in the produce method

gcn_kafka/core.py

@@ -11,6 +11,9 @@
 import certifi
 import confluent_kafka
 import confluent_kafka.admin
+from jsonschema import validate, exceptions
+import json
+import requests
 
 from .oidc import set_oauth_cb
 
@@ -91,6 +94,25 @@ def __init__(
         self.poll(0)
 
 
+    def produce_json_notice(self, topic:str, data:dict):
+        try:
+            request = requests.get(data['$schema'])
+            schema_model = json.loads(request.content.decode())
+            validate(data, schema_model)
+        except exceptions.ValidationError:
+            print("The instance of the data provided is invalid against the schema")
+            return
+        except exceptions.SchemaError:
+            print("The provided schema is invalid")
+            return
+
+        notice_id = str(uuid4())
+        data['notice_id'] = notice_id
+
+        super().produce(topic, json.dumps(data))
+        print(f'Successfully posted topic with notice id: {notice_id}')
+
+
 class Consumer(confluent_kafka.Consumer):
     def __init__(
         self,

pyproject.toml

@@ -24,7 +24,8 @@ dependencies = [
     "certifi",
     "confluent-kafka >= 1.6.1",
     "requests",
-    "typing-extensions; python_version<='3.7'"
+    "typing-extensions; python_version<='3.7'",
+    "jsonschema"
 ]
 requires-python = ">=3.7"
 dynamic = [ "version" ]