Stream Your Data Using Nothing But Python’s Requests Library

The Forgotten Streaming Library Hidden In Python

When you think of building streaming data pipelines you likely think of using a library like Apache Kafka that is tailor-made for streaming.

Cloud platforms like Google Cloud Platforms (GCP) and Amazon Web Services (AWS) support the streaming of data into their respective SQL environments. GCP, in particular, even provides services like the Billing API or Google Analytics connections that stream source data on a recurring basis.

On top of the infrastructure giants countless third-party “code-less” platforms have taken on the heavy lifting of setting up, maintaining and scaling streaming infrastructure.

Even though third parties sell platforms that will alleviate the “challenging” builds like streaming pipelines, as a data engineer, it is still important to know how to configure a streaming pipeline.

But going from creating relatively simple batch load jobs to configuring dynamic streaming pipelines can be, understandably, intimidating.

Aside from not understanding the concept of streaming data, it can be difficult to even know where to begin when creating such pipelines.

Enter your Fisher-Price “My First Streaming Pipeline”: Requests.

Build Your Pipeline To A Data Engineering Career

You’ve reached the limit of the public preview. The full version of this post includes the implementation details: The code, the edge cases, and the "why" behind the architecture.

When you join PipelineToDE, you get:

• The DA → DE Pathway Course: A structured roadmap to bridge the gap between analysis and engineering.
• Weekly Senior Deep Dives: Fresh, tactical insights on Python, Cloud (GCP/AWS), and modern orchestration delivered every week.
• Production-Ready Blueprints: Access to 80+ protected stories and code repos from my time in the trenches as a Senior DE
• The DE Job Board (Coming Soon): Exclusive access to a curated board of high-agency Data Engineering roles.

I Built My First Streaming Pipeline By Accident

At work I was working on a project that involved creating a relatively simple ETL batch pipeline that ran on a recurring basis.

Initially working with test data, I had not tested the rigors of the API against larger volumes of data.

So, when we switched to production and wanted to do an initial import from the API… it broke. Badly.

After investigating, I discovered that while the test data was a small subset of the data we initially needed and could be retrieved with a standard request —

req = requests.get(url, token, auth) 
json_response = req.json()

However, when it came time to GET the prod data, the way the data was coming in was entirely different, which required me to rip apart my initial build.

What’s In Your Header

When I‘m troubleshooting an issue with a data pipeline I typically check two things:

• The status code (200, 400, 429, etc.)
• The text payload returned

One thing I’ve neglected, until this point, is to check the header returned by the API.

However, the header, typically accessed by the “.header” method, can provide critical context for how you need to configure your pipeline to properly receive the data returned.

One of the most telling way to determine what kind of protocol you’re dealing with is by examining the “Content-Length” in the header returned.

If you see a content length parameter, you can be assured that most likely your request will return all the data, all at once, since the length is predetermined.

However, if you don’t see a content length parameter, this is a good indication you are dealing with dynamic (streaming) data, that you will need to receive in intervals or “chunks” when using the requests library.

A tell-tale sign is the HTTP protocol listed in the header will be “TransferEncoding” rather than Content Encoding.

Typically the header will be “TransferEncoding: chunked.”

Think of the difference between these two configurations as the difference between a defined list and a list you have to iterate over and append onto.

In the first list, you know exactly how long your list is without having to do any kind of read/write operation. In the second list, you won’t know how long the list is until the operation terminates.

Before you can even think about configuring a pipeline, you need to master the difference between these two protocols.

Streaming “Chunks” Of Data In Requests

Transfer Encoding: chunked is an HTTP protocol that allows data of undetermined length to be sent to your pipeline in “chunks” rather than in one definitively-sized batch.

This helps break up large files (100s of MB or even > 1 GB).

Chunks will typically be sent in an encoded bytes format.

Unlike other bytes formats you may see, bytes read in a TransferEncoding process are formatted a bit differently.

In addition to the byte text, the chunk number precedes the byte payload.

For example (forgive the alphanumeric characters; I’m human, after all).

0 /n[aasdf]/n

But before worrying about decoding the bytes, we need to get them.

To enable streaming in “requests” you just need to add the “stream” parameter.

req = requests.get(url, token, auth, stream=True)

As long as stream=True, you‘ll receive data in a streaming form.

However, leaving this connection open is a bad practice.

To stabilize this request, you’ll want to wrap it in a with statement:

with requests.get(url, token, auth, stream=True) as r:

Adding the with statement ensures that your connection will remain open as long as data is being streamed from the API.

To access the chunks that are returned you’ll need to create a simple loop.

You can access chunks of a data stream by using either iter_content (which allows for a chunk_size to be specified) or iter_lines.

with requests.get(url, token, auth, stream=True) as r: 
  for chunk in r.iter_content(chunk_size=100): 
      print(chunk)

with requests.get(url, token, auth, stream=True) as r: 
  for chunk in r.iter_lines(): 
      print(chunk)

Working with bytes can be particularly tricky because, like any other data source, they can contain “bad data” or, in TransferEncoding, a “bad chunk.”

To prevent this, we need to think about creating a check to eliminate bad chunks. A simple if condition will fulfill this use case.

with requests.get(url, token, auth, stream=True) as r: 
  for chunk in r.iter_lines(): 
      if chunk: 
            print(chunk) 
      else: 
          logging.info("No chunks returned...")

Once we’re sure we’re capturing “good” chunks, it’s time to store the output.

Buffering and Writing

Many resources surrounding TransferEncoding suggest adding a buffer as additional protection against bad chunks.

By “buffer”, I simply mean adding a fail-safe in the process.

One of the best ways to do this is to write to a file of your specified output.

Before we can do that, however, we have to specify the format of the final data returned.

If I want a JSON file, I’ll want to specify this in the open() argument.

To decode the bytes being streamed, I can use the .decode method.

with requests.get(url, token, auth, stream=True) as r: 
  with open("output.json", "w") as f: 
     for chunk in r.iter_lines(): 
         if chunk: 
              chunk = chunk.decode("utf-8") 
              f.write(chunk) 
         else: 
              logging.info("No data returned.")

The last step, to make this a durable request, is to add exception handling.

Specifically, we’ll want to except the ChunkedEncoding error that I’ve encountered way too many times.

from requests.exceptions import ChunkedEncodingError as ChunkedEncodingError 
 
try: 
   with requests.get(url, token, auth, stream=True) as r: 
    with open("output.json", "w") as f: 
       for chunk in r.iter_lines(): 
           if chunk: 
                chunk = chunk.decode("utf-8") 
                f.write(chunk) 
           else: 
                logging.info("No data returned.") 
 
except ChunkedEncodingError as e: 
      logging.info(f"Error detected - {e}")

Takeaway

The snippets shared in this overview constitute the “meat” of any streaming pipeline.

When I was first building a pipeline using the “requests” library’s stream=True parameter, the most challenging part of my build was figuring out how to configure a chunking process.

Adjusting a pipeline, especially on the fly (in my case) is never easy.

In addition to being aware of status codes and the payloads being returned, you’ll want to understand the “ideal” way of communicating with the API by accessing its headers.

When a header says TransferEncoding: chunked, that’s when you know you’ll want to re-work your batch pipeline to handle streaming data.

The most important takeaway I gleaned from my process was that, when it comes to APIs you can’t just make calls. You need to communicate with it.