Skip to content

Polling a REST API service

If there's an existing REST API you want to pull data from, you can write a Quix connector that polls the REST API. By way of example, the following code demonstrates the idea:

from quixstreams import Application
from quixstreams.models.serializers.quix import JSONSerializer, SerializationContext
import time, os, requests

app = Application()

serializer = JSONSerializer()
output_topic = app.topic(os.environ["output"])
producer = app.get_producer()

while True:
    response = requests.get("")
    json_response = response.json()

    with producer:
        serialized_value = serializer(
            value=json_response, ctx=SerializationContext(

The code requests data from the REST API and then publishes it to a stream.