Produce Coingecko Recorrds to Kafka

coingecko-to-kafka.py

import requests
import datetime
from kafka import KafkaProducer
from json import dumps
from kafka.errors import KafkaError


def produce():
    producer = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda x: dumps(x).encode('utf-8'))
    response = requests.get('https://api.coingecko.com/api/v3/exchange_rates').json()['rates']
    result = [dict(response[k], **{'timestamp': datetime.datetime.now().isoformat()}) for k in response]
    for msg in result:
        future = producer.send('druid-sample-data', value=msg)
        try:
            future.get(timeout=10)
        except KafkaError:
            print('failed')
            pass


if __name__ == '__main__':
    produce()