Mastering the Easy Art of Delegation in Python: A Step-by-Step Guide

Introduction

Delegation is like passing a job to someone else. For example, while performing a download from a file from the web, a delegate could handle the events, like starting the download, maintaining updates on the download’s progress, and handling the download when it is finished. Also, and this is quite important, it can handle errors.

Why do it this way? Using a delegate can make the code more flexible. Also, using a delegate promotes the separation of concerns.

Implementation in Python

In this example we will build a simple file downloader, which can download a file from a URL. In this example we will use the requests package to download the file, so install that first and type this in your terminal or command line:

pip install requests

Next import the needed modules:

import requests
import os

Why do we need these? The requests package is needed to facilitate making web requests, and since we want to write the downloaded file to a file system, we also need the os package.

The delegate interface

We want flexibility, so let’s make sure the delegates for handling the downloading events all have the same interface:

class DownloaderDelegate:
    def start(self):
        pass
    def progress(self, current, total):
        pass
    def end(self):
        pass
    def error(self, message):
        pass

The delegate defines four methods:

1. The start() method which should be called when the download starts.
2. A progress() method, which is updated as the download progresses.
3. An end() method which handles the logic when the download is finished.
4. And finally error() should be called when an error occurs.

The downloader class

The file downloader class has two methods, and we will start with the constructor:

    def __init__(self, delegate: DownloaderDelegate | None = None):
        self._delegate = delegate

This simply sets the delegate, or not, since the delegate can be None which means no delegation will take place.

The actual download method has a bit more code:

    def download(self, url: str, destination: str):

        if self._delegate:
            self._delegate.start()
        
        try:
            response = requests.get(url, stream=True)
            response.raise_for_status()

            total_size = int(response.headers.get('content-length', 0))
            if total_size == 0:
                raise ValueError("File size is zero.")

            with open(destination, 'wb') as file:
                downloaded_size = 0
                for data in response.iter_content(chunk_size=128):
                    file.write(data)
                    downloaded_size += len(data)
                    if self._delegate:
                        self._delegate.progress(downloaded_size, total_size)

            if self._delegate:
                self._delegate.end()
        except Exception as e:
            if self._delegate:
                self._delegate.error(str(e))

Line by line:

1. The download() method receives two parameters (apart from self): the url from which to download the content, and the file to which to write the content.
2. We first start by checking if there is a delegate, and if so, call its start() method.
3. In the try-catch block we start by trying to download from the supplied url.
4. If an error code is return, e.g. a 404 or a 500, the raise_for_status() method raises and exception, and the error can be handled.
5. Next we check the size of the file, if it is zero, an exception is raised.
6. Next we open our local file, in write binary mode. This because we can download text files from an url, we could as well download binary files like images or zip files.
7. We set the downloaded size to zero.
8. Next, using the iter_content() method we download chunks from the file, in our case 128 bytes each, write it to our file, update the downloaded size, and call the progress() method on the delegate.
9. When the download is done, we call the end() method on the delegate.
10. Finally, if an exception occurred, we call the error() method on the delegate.

The delegate implementation

Now we come to the implementation of the delegate itself. One thing to note is that the DownloaderDelegate, which is the interface is now the parent of the concrete DownloadDelegate class. The methods in this class are nothing more than simple print statements:

class DownloadDelegate(DownloaderDelegate):
    def start(self):
        print("Download started.")
    
    def progress(self, current, total):
        print(f"Downloaded {current} of {total} bytes.")
    
    def end(self):
        print("Download completed successfully.")
    
    def error(self, message):
        print(f"Error occurred: {message}")

Testing time

Now we can test this simple setup:

if __name__ == "__main__":
    url = "<file url>"
    destination = "<filename>"

    delegate = DownloadDelegate()
    downloader = FileDownloader(delegate)
    
    if not os.path.exists(destination):
        downloader.download(url, destination)
    else:
        print(f"File already exists at {destination}.")

Note: replace the “<file url>” and “<filename>” values with your own.

Line by line:

1. Start by defining the URL from where the file is to be downloaded, and the local filename.
2. Instantiate a delegate, please keep in mind that this should be the concrete class.
3. Use this delegate to create a new downloader.
4. Finally check if the file already exists, if so, print a message and exit the program.
5. If not, start the download and see the messages appear.

Conclusion

As you can see, delegation can be quite powerful, and add some flexibility to your code. Possible extensions would be to have more than one delegate, so a series of delegates would be called when for example the download progresses.

Another possible extension would be to make this code more thread safe, but that will be the subject of another article.