def publish_messages()

in pipelines/iot_analytics/scripts/publish_on_pubsub.py [0:0]

• 12 lines of code
• 3 McCabe index (conditional complexity)

def publish_messages(project, topic, data_path):
  """
    Publishes JSON messages from a file to a Pub/Sub topic.

    Args:
        project: The ID of the Google Cloud project.
        topic: The ID of the Pub/Sub topic.
        data_path: The path to the JSON data file.
    """

  publisher = pubsub_v1.PublisherClient()
  topic_path = publisher.topic_path(project, topic)

  with open(data_path, "r", encoding="utf-8") as f:
    for line in f:
      try:
        # Parse each line as a JSON object
        json_data = json.loads(line)

        # Publish the JSON data as a message
        message_data = json.dumps(json_data).encode("utf-8")
        future = publisher.publish(topic_path, message_data)
        print(f"Published message ID: {future.result()}")

      except json.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")