1 files changed, 34 insertions, 0 deletions
diff --git a/python/fatcat_web/kafka.py b/python/fatcat_web/kafka.py
new file mode 100644
index 00000000..895e719f
--- /dev/null
+++ b/python/fatcat_web/kafka.py
@@ -0,0 +1,34 @@
+
+import requests
+
+from fatcat_web import Config
+
+
+def kafka_pixy_produce(topic, msg, key=None, sync=True, timeout=5):
+    """
+    Simple helper to public a message to the given Kafka topic, via the
+    configured kafka-pixy HTTP gateway
+
+    topic: string
+    msg: string
+    key: optional, bytes
+    timeout: seconds
+    """
+
+    if not Config.KAFKA_PIXY_ENDPOINT:
+        raise Exception("Kafka produce error: kafka-pixy endpoint not configured")
+
+    params = dict()
+    if key:
+        params['key'] = key
+    if sync:
+        params['sync'] = True
+    resp = requests.post(
+        "{}/topics/{}/messages".format(Config.KAFKA_PIXY_ENDPOINT, topic),
+        params=params,
+        data=msg,
+        headers={"Content-Type": "text/plain"},
+        timeout=timeout,
+    )
+    resp.raise_for_status()
+    print(resp.json())