Publish data

This page explains how to publish data to GlassFlow pipelines.

The Python SDK provides a programmatic way to interact with GlassFlow pipelines and the library is used to publish or consume data.

Create a data publisher for the mobility project

In this section, we'll explore how to generate data for the mobility project and publish it to the data pipeline in GlassFlow using the Python SDK. We created a data pipeline with a transformation function in the previous section.

Prerequisites

We assume that you have already completed the following before proceeding with the tutorial:

• You completed the Create a Pipeline tutorial.

Install required libraries

Install required libraries including GlassFlow SDK listed in the requirements.txt file using the pipcommand in a terminal.

pip install -r requirements.txt

Publish real-time API events to the pipeline

Create a new Python script file called producer_api.py in your project root directory and insert the below code. This Python script serves as a data producer, fetching mobility events data from a mock API server and publishing it to a GlassFlow pipeline.

https://github.com/glassflow/glassflow-examples/blob/main/tutorials/mobility/producer_api.py

"""
Get mobility events data via a mockserver and publish it to glassflow
"""
import glassflow
from dotenv import dotenv_values
import requests


def get_mock_events():
    """
    Get mock events from the mock server
    """
    res = requests.get(
        "https://mock-mobility-s3r3lbzina-ey.a.run.app/mobility/producer/events/ride-completed"
    )
    if res.status_code == 200:
        return res.json()
    else:
        print("Failed to get mock events")
        return None


def main():
    config = dotenv_values(".env")
    pipeline_id = config.get("PIPELINE_ID")
    space_id = config.get("SPACE_ID")
    token = config.get("PIPELINE_ACCESS_TOKEN")

    client = glassflow.GlassFlowClient()
    pipeline_client = client.pipeline_client(space_id=space_id,
                                             pipeline_id=pipeline_id,
                                             pipeline_access_token=token)
    counter = 0
    while True and counter < 1000:
        try:
            event = get_mock_events()
            counter += 1
            if event:
                req = pipeline_client.publish(request_body=event[0])

                if req.status_code == 200:
                    print("Event published successfully", event[0])
                else:
                    print("Failed to publish event")
                    print(req.text)
        except Exception as e:
            print(e)
            break
        except KeyboardInterrupt:
            break


if __name__ == "__main__":
    main()

In the above Python code, there are a few things to consider like initializing the GlassFlow client, interacting with the pipeline, and publishing data to the pipeline.

1. Initializes a client for interacting with the GlassFlow platform. It provides access to various functionalities, including pipeline management, data publishing, and data pulling.

client = glassflow.GlassFlowClient()

2. Creates a pipeline client object specifically for interacting with the particular pipeline within a specified space. It takes three parameters: space_id (the ID of the space where the pipeline resides) , pipeline_id (the ID of the pipeline) and pipeline_access_token (the access token required for authentication). It returns a client object that can be used to publish data to the specified pipeline.

pipeline_client = client.pipeline_client(space_id=space_id, 
                                         pipeline_id=pipeline_id, 
                                         pipeline_access_token=token)

3. Publishes a new event into the pipeline. It takes one parameter: request_body (the data to be published) . Under the hood, it sends a request to the GlassFlow API endpoint with the provided data to publish the new event to the given pipeline.

pipeline_client.publish(request_body)

Refer to Python SDK documentation for more details.

Run the script

Run the Python script producer_api.py

python producer_api.py

This script continuously fetches mock mobility events data from a mock API server and publishes it to the specified GlassFlow pipeline.

Next

After successfully producing, and publishing data into the pipeline, and transforming it, you can now consume the transformed data.