code: def publish(self, topic: str, records: [RideRecordKey, RideRecord]):
      for key_value in records:
          key, value = key_value
          try:
              self.producer.produce(
                  topic=topic,
                  key=self.key_serializer(
                      key,
                      SerializationContext(
                          topic=topic, field=MessageField.KEY
                      ),
                  ),
                  value=self.value_serializer(
                      value,
                      SerializationContext(
                          topic=topic, field=MessageField.VALUE
                      ),
                  ),
                  on_delivery=delivery_report,
              )
          except KeyboardInterrupt:
              break
          except Exception as e:
              print(f"Exception while producing record - {value}: {e}")
  
      self.producer.flush()
      sleep(1)
  
file_path: data-engineering-zoomcamp/06-streaming/python/avro_example/producer.py
imports: import os
  import csv
  from time import sleep
  from typing import Dict
  from confluent_kafka import Producer
  from confluent_kafka.schema_registry import SchemaRegistryClient
  from confluent_kafka.schema_registry.avro import AvroSerializer
  from confluent_kafka.serialization import SerializationContext, MessageField
  from ride_record_key import RideRecordKey, ride_record_key_to_dict
  from ride_record import RideRecord, ride_record_to_dict
  from settings import RIDE_KEY_SCHEMA_PATH, RIDE_VALUE_SCHEMA_PATH, SCHEMA_REGISTRY_URL, BOOTSTRAP_SERVERS, INPUT_DATA_PATH, KAFKA_TOPIC
label_desc: Use image node_id 5 for calling the RideAvroProducer obj's underlying member method code with example usage: obj.publish(topic, records) without return types