Python - streaming file

Question

I need to download and upload around 1000 files. The file is first downloaded from one webserver and then uploaded to another. My first approach was to save each file on disk and then upload - but as I spend around 30GB combined for all files, I decided to look into streaming each file from the first webserver to the second so I avoid writing to disk. I have some working code, but I'm unsure if this is done right. Would this be the right approach to solving this problem?

import requests
from requests.auth import HTTPBasicAuth
from io import BytesIO

def stream_file(source_url):
  with requests.get(source_url, headers={"Authorization": "Bearer abc"}, stream=True) as source_response:
    if source_response.status_code != 200:
        print("Error accessing the source file:", source_response.status_code)
        return

    print(source_response.status_code)
    # Accumulate streamed content into a BytesIO object
    content_buffer = BytesIO()
    for chunk in source_response.iter_content(chunk_size=8192):
        content_buffer.write(chunk)
    
    # Pass the BytesIO object to the upload_file function
    upload_file(content_buffer)    
    print("File streamed successfully.")

def upload_file(file_stream):
  nexus_url = "http://webserver"
  nexus_username = "admin"
  nexus_password = "admin"
  upload_url = f"{nexus_url}/service/rest/v1/components?repository=test"
  auth = HTTPBasicAuth(nexus_username, nexus_password)

  payload = {
      'raw.directory': (None, '/'),
      'raw.asset1.filename': (None, "file.gz"),
      'raw.asset1': ("file.gz", file_stream.getvalue())
  }

  print(f"Uploading stream to Nexus")
  response = requests.post(upload_url, files=payload, auth=auth)
  print(response.text)

# Example usage
source_artifactory_url = "https://artifactory.file.gz"

stream_file(source_artifactory_url)

Answer 1

A small optimization you could do is to use session and you can expect a performance boost from that. You can also shorten the code a little bit by setting auth at session level:

import requests
session = requests.Session()
session.auth = ('user', 'pass')
url = "..."
response = session.post(upload_url, files=payload)

Config options should preferably be set at the top of your module, in uppercase if they are constants. But for sensitive values like username and password a better option would be to use .env files or another secrets management solution. Because you don't want to commit credentials in your SCM (git/other) repo.

Probably you could run a few threads in parallel to speed up code execution. The ThreadPool class would be a suitable option. Example: A Threadsafe Connection Pool for Requests

Don't forget to check the status code after sending your post request. There is no exception handling, at a minimum you would want to use built-in exceptions provided by the requests module (requests.exceptions). Function stream_file returns nothing useful in case of error. You could simply raise an exception (using raise_for_status) if the status code is != 200. In any case, your application should either retry or stop, but not ignore failure and continue blindly.

If this code is meant to be used as a CLI tool, I would love to see some kind of progress bar and the tqdm package is there for that. If there is a lot of large files, this is going to take some time and it is nice to be able to track progress and even calculate an approximate ETA.

When downloading files, the Content-Length header will tell you the size in advance, but that header is not always present. Its value may also be -1 if not available, so only positive values should be taken into consideration.

Would this be the right approach to solving this problem?

I don't know if you are dealing with some kind of third-party API that you "must" use, or if you are transferring files between two servers that are both under your control. In that latter case, a better solution would be to use rsync. That would be easier and there would be less overhead than using an HTTP-based solution. rsync may already be installed on your Linux machine, SCP too.

Compressing the files prior to transfer may be a smart tactic depending on the file type. If you are moving JPEG images, the gain is virtually nil. If you are transferring text files, log files etc the file size is going to be substantially reduced, resulting in accelerated transfers.

Answer 2

This isn't "streaming". Each gigabyte input file will fill up a gigabyte of RAM before even a single byte of it is sent to the destination server.

With type annotation, the signature would be

def stream_file(source_url: str) -> None:

Or roughly equivalently we might see

def stream_file(source_url: str) -> BytesIO:

with calling sequence of upload(stream_file(source_artifactory_url))

Here is the OP loop:

    content_buffer = BytesIO()
    for chunk in source_response.iter_content(chunk_size=size):
        content_buffer.write(chunk)

streaming

To achieve your stated goal, you'll want to interleave reads with writes.

    uploader = get_uploader(dest_url)

    for chunk in source_response.iter_content(chunk_size=size):
        content_buffer = BytesIO()
        content_buffer.write(chunk)
        uploader.send_chunk(content_buffer)

Of course, content_buffer is no longer serving any useful role at this point, so we could elide it and simply .send_chunk(chunk). And you'll want to write that get_uploader() helper.

All of that sets us up for this revised signature:

def stream_file(source_url: str, dest_url: str) -> None:

We're back to evaluating for side effects, returning nothing. But this time we're streaming! Good.

advantages

network behavior

This "copy chunks from src to dst" loop lets us finish writing to the destination's disk with lowest possible latency. You probably care less about that than throughput. It's also good for throughput, as it lets us keep the pipe busy in both directions.

Which brings us to the notion of "bottleneck". With three machines, two filesystems, and two end-to-end TCP connections, it's unclear whether we run out of CPU cycles somewhere, or disk I/O bandwidth on {src, dst} filesystem, or TCP network bandwidth for {down,up}loading. You will want to make measurements so you better understand that. Doubling the disk bandwidth won't make any difference at all if you're routinely maxing out some network link.

The next step on your journey would be to turn this simple 2-connection loop into one which manages 2 × N connections, so you have N file transfers happening simultaneously. They could involve the same pair of (very powerful?) source and destination servers, but could also involve up to N distinct source and N distinct destination servers. Which would let you throw lots more hardware resource at the problem and perhaps finish in an hour rather than six hours, assuming you've lots of network bandwidth.

memory behavior

Any collection of input files is likely to be comprised of different sizes, often with a Zipfian distribution. If you do a thousand allocate-followed-by-free operations, of different sizes, your memory footprint keeps ratcheting up until it hits max( … ) of all those sizes.

Sending each chunk as soon as it arrives reduces local resource needs down to \$O(1)\$ constant memory complexity, independent of max size.

disadvantages

If you're going to be transferring file(s) for a few minutes, then you're sitting on resources on both {src, dst} servers, which can affect other users of those servers. Suppose one of the two servers is less powerful and is therefore the bottleneck. In the OP non-streaming case the servers are decoupled, and can finish a {down,up}load task at their maximum speed, then devote resources to rapidly serving other users. When streaming, the more powerful one will need to commit resources for a longer time.

Likely either one of {src, dst} server (with network connection) will be the bottleneck. Bumping up the concurrent sessions to 2 or 3 might increase observed system throughput (byte/second), beyond which at some point additional sessions won't improve matters. But the extra overhead from those additional sessions can impact the response seen by you or by other users. So "too much concurrency" can be a net lose. The OP code supports just a single session, so it is fundamentally unable to produce such lossage. Take care when tuning the concurrency level, informed by observed performance figures.

buffer size

That chunk_size=8192 impresses me as being on the small side. We perform a small 8 KiB chunk of work in the TCP stack, and then have to wait for cPython to interpret some bytecode, and then move on to the next small chunk of work.

Better to use bigger chunks, big enough that time spent interpreting bytecode won't be the dominant source of delay. Use profiling measurements to better understand the tradeoffs at different buffer sizes.