kproducer.py.save

from kafka import KafkaProducer
from datetime import datetime
from json import dumps
from random import random, choice
from helper import Purchasse,  Bookmark, Comment, Basket, Click, Generator, APP_NAME, KAFKA_BOOTSTRAP_SERVERS, ACTIONS, C, W
from time import sleep
from sys import stdout

producer = KafkaProducer(
    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS, 
    value_serializer=lambda x: dumps(x).encode('utf-8')
)


(
   lambda name, rule_size: stdout.write(
      f"{'='*(len(name)+rule_size)}"
      +"\n"
      +f"|{name:^{len(name)+(rule_size-2)}}|"
      +"\n"
      +f"{'='*(len(name)+rule_size)}"
      +"\n"
   )
)(APP_NAME, 10)


def produce_event():
    while True:
        event = Generator.emit() 
        producer.send(topic=event.action, value=event.__dict__)
        produce_event.count += 1
        # stdout looks good flush the buffer
        stdout.write(str(event))
        stdout.write("event_count " + C + " % " + str(produce_event.count) + W)
        sleep(random()) 
    
produce_event.count = 0
produce_event()