fbpx
Why PEP8 is not enough to write Pythonic code (2/2)

This article is the fifth part of the entire entry. If you haven’t read the first part, look here and let us know if it is helpful to you!

Context manager

Imagine this situation: you are serving an empire as a Senior Developer in a project to develop and maintain software that communicates and manages probes that are used to comb through wreckages of spaceships destroyed during space battles. This project is quite important as battleships are full of interesting artifacts. They are also very useful to make sure that there are no survivors on those ships.

Part 1 – Preludium

Your project just got a new developer. For his service to the Empire, he was transferred to your project (after all it a prestigious position). He worked as a C++ developer writing firmware for Hyperdrives (every freshman can do this!). He is new, you want to go easy on him so you give him a simple task: write python adapter for C code that is used to retrieve data from probes [this library supports two types of probes with different data formats and communication APIs – one for probes that work on a planets surface and another one for those working in outer space]. You want adapter because it’s a C code – you have return values, you can cause memory errors if you don’t initialize this library properly etc. It’s a simple task and you don’t want to be responsible for even a minute of time wasted while serving The Emperor so you also ask him to write a script that uses aforementioned adapter class and sends data to analyze() function prepared by Empire Looting Division

We already have some code created for WDP project. Luckily for us (and our new team member), it’s not that much so we can put it in here:

		
def analyze(value):
    """Very complicated function and not made up example one"""
    print("Analyzing data from probe: ", value)
    print("Analyze complete for probe ", value)


class Logging:
    """Distributed logging system with data retention plan - obviously"""

    @staticmethod
    def log_err(value):
        print("Error: " + value)

    @staticmethod
    def log_info(value):
        print("Info: " + value)


class UpdateFirmwareException(Exception):
    pass


class UpdateFirmware:
    def update_456(self, addr):
        pass

    def update(self, addr):
        pass
		
	

After a few hours of intensive work, he showed us the results of his work.Adapter
WreckageDebrisProbes.py:

		
import c_library_to_retreive_data_from_probes as iprobe

class WreckageDebrisProbesException(Exception):
    pass

class WreckageDebrisProbes:
    def __init__(self, bind_addr):
        self.address = bind_addr
        self.binded = False
        self.iprobe = iprobe()

    def get_space_probes(self):
        if not self.binded:
            raise WreckageDebrisProbesException
        return self.iprobe.get_space_probes()

    def get_ground_probes(self):
        if not self.binded:
            raise WreckageDebrisProbesException
        return self.iprobe.get_ground_probes()

    def bind(self):
        self.binded = self.iprobe.bind(self.address)
        if self.binded:
            Logging.log_info("Binded to interface")
        else:
            Logging.log_err("Binding to interface failed")

    def unbind(self):
        self.iprobe.unbind()
        self.binded = False
        Logging.log_info("Unbinded from interface")
		
	

Script that sends data to analyze() function
SendToAnalyze.py:

		
#!/usr/bin/python
from WreckageDebrisProbes import WreckageDebrisProbes
from WreckageDebrisProbes import WreckageDebrisProbesException

try:
    # yep, we are still using IPv4 in XXII century
    conn_devices = WreckageDebrisProbes("192.168.0.1")
    conn_devices.bind()
    probes = conn_devices.get_space_probes()
except WreckageDebrisProbesException:
    Logging.log_err("Errors while trying to scan local network")
else:
    for probe in probes:
        analyze(probe)
    Logging.log_info("Analyze complete")
finally:
    conn_devices.unbind()
		
	

Everything looks fine, it’s mostly acceptable PEP8 code, he even took care of logging – all good in the hood. But something is a little bit off. It smells a little bit like C++. But it’s pep8 so it has to be good.

Part 2 – Climax

Months go by and suddenly you got information from Empire Cyber Security Division (ECSD) that some of your ground probes (Class 3 and higher) require a firmware upgrade. But it’s even worse. Depending on which version of firmware probe have you might have to upgrade to different firmware BEFORE upgrading to the proper one (something about using “twos” instead of “ones” and “zeros” as bits). It seems that problematic firmware has version number equal to 456. We have to update it asap. Of course, we don’t keep data about probes (spies* etc.) so we have to write a code that:

– Get information from probes about their addresses and firmware versions from broadcast;

– Use retrieved information to upgrade affected probes

It should not be that hard. We have this adapter written a few months ago – it should be able to handle most of the work. We also have a function to update probes. It would make sense to ask the author of the adapter to write update script but unfortunately, he was executed a few months ago for using twos instead of ones and zeros in binary code (wait, why that sound familiar?). Fortunately, you have a new junior developer who seems to understand this stuff so you ask him to write a script.

After a few hours, you get something like that from him:

		
#!/usr/bin/python
from WreckageDebrisProbes import WreckageDebrisProbes
from WreckageDebrisProbes import WreckageDebrisProbesException

try:
    conn_devices = WreckageDebrisProbes("192.168.0.1")
    conn_devices.bind()
    probes = conn_devices.get_ground_probes()
except WreckageDebrisProbesException:
    Logging.log_err("Errors while trying to scan local network")
else:
    up_firmware = UpdateFirmware()
    for probe in probes:
        if probe['class'] >= 3:
            try:
                if probe['firmware_version'] == 456:
                    up_firmware.update_456(probe['address'])
                    Logging.log_info("Intermediate update complete for: " +
                                     probe['address'] +
                                     "[" + probe['name'] + "]")
                up_firmware.update(probe['address'])
                Logging.log_info("Update complete for: " +
                                 probe['address'] +
                                 "[" + probe['name'] + "]")
            except UpdateFirmwareException:
                Logging.log_err("Update FAILED for: " +
                                probe['address'] +
                                "[" + probe['name'] + "]")
finally:
    conn_devices.unbind()
		
	

You verified the code in the testing environment, seems to be working, it’s pep8 so you run it in production. Everything went fine:

Info: Binded to interface
Info: Update complete for: 192.168.0.2[Doc]
Info: Intermediate update complete for: 192.168.0.3[Bashful]
Info: Update complete for: 192.168.0.3[Bashful]
Info: Intermediate update complete for: 192.168.0.5[Grumpy]
Info: Update complete for: 192.168.0.5[Grumpy]
Info: Update complete for: 192.168.0.6[Sleepy]
Info: Unbinded from interface
		

Life is good, you can enjoy the rest of your day, right? Wrong. Code you accepted was not pythonic you feel horrible so you start thinking about how to improve it. Even though it’s already pep8 compliant [Ominous noises in the background].

*After a few years, we found out that our enemy had a list of our ground probes. Here it is:

{"name": "Dopey", "class": 2, "firmware_version": 279, "address": "192.168.0.1"},
{"name": "Doc", "class": 3, "firmware_version": 738, "address": "192.168.0.2"},
{"name": "Bashful", "class": 5, "firmware_version": 456, "address": "192.168.0.3"},
{"name": "Happy", "class": 2, "firmware_version": 345, "address": "192.168.0.4"},
{"name": "Grumpy", "class": 3, "firmware_version": 456, "address": "192.168.0.5"},
{"name": "Sleepy", "class": 7, "firmware_version": 25, "address": "192.168.0.6"},
{"name": "Sneezy", "class": 2, "firmware_version": 1345, "address": "192.168.0.7"}
		

Part 3 – Enlightenment

After 30 seconds of thinking and a few days of playing computer games, you realize what is the problem. Code you accepted is not elegant. Every time someone is using WreckageDebrisProbes class they have to take care of errors themselves. Code is very repetitive and boils down to the fact that try, except and finally blocks are very similar for every usage. You basically:

  1. Initialize library with proper address
  2. Bind interface
    • Goto 5 if binding failed
  3. Retrieve data/information from probes
  4. Do something with that information
  5. Clean-up

In other words it always looks like this:

		
try:
    conn_devices = WreckageDebrisProbes("192.168.0.1")
    conn_devices.bind()
    probes = conn_devices.get_space_probes() # or other types of probes supported by WreckageDebrisProbes
except WreckageDebrisProbesException:
    Logging.log_err("Errors while trying to scan local network")
else:
	# DO INTERESTING STUFF
finally:
    conn_devices.unbind()
		
	

It’s a lot of boilerplate code. Wouldn’t it be great if you could push interface management onto WreckageDebrisProbes and use it without knowing WreckageDebrisProbes insides?

In Python you can!
We can use something called a context manager. Have you heard about “with”? You can implement it for your class and make sure that people using your class will concentrate on things that are important.

Let’s improve WreckageDebrisProbes!

All we have to do is to implement __enter__ and __exit__ methods and modify our code responsible for raising exceptions when there is nothing binded:

		
class WreckageDebrisProbes:
    def __init__(self, bind_addr):
        self.address = bind_addr
        self.binded = False
        self.iprobe = iprobe()

    def __enter__(self):
        self.bind()
        return self

    def __exit__(self, exctype, value, traceback):
        if exctype == WreckageDebrisProbesException:
            Logging.log_err("Errors while trying to scan local network")
        self.unbind()

    def get_space_probes(self):
        if not self.binded:
            return []
        return self.iprobe.get_space_probes()

    def get_ground_probes(self):
        if not self.binded:
            return []
        return self.iprobe.get_ground_probes()

    def bind(self):
        self.binded = self.iprobe.bind(self.address)
        if self.binded:
            Logging.log_info("Binded to interface")
        else:
            Logging.log_err("Binding to interface failed")

    def unbind(self):
        self.iprobe.unbind()
        self.binded = False
        Logging.log_info("Unbinded from interface")
		
	

Now our code to get data and send to analysis would look like that:

		
with WreckageDebrisProbes2("192.168.0.1") as wdp:
    probes = wdp.get_space_probes()
    for probe in probes:
        analyze(probe)
    Logging.log_info("Analyze complete")
		
	

And code to update firmware:

		
with WreckageDebrisProbes("192.168.0.1") as wdp:
    probes = wdp.get_ground_probes()

    up_firmware = UpdateFirmware()
    for probe in probes:
        if probe['class'] >= 3:
            try:
                if probe['firmware_version'] == 456:
                    up_firmware.update_456(probe['address'])
                    Logging.log_info(
                        "Intermediate update complete for: " + probe['address'] + "[" + probe['name'] + "]")
                up_firmware.update(probe['address'])
                Logging.log_info("Update complete for: " + probe['address'] + "[" + probe['name'] + "]")
            except UpdateFirmwareException:
                Logging.log_err("Update FAILED for: " + probe['address'] + "[" + probe['name'] + "]")
		
	

Quite an improvement, isn’t it? And we still have some things to do to make this code more pythonic…

Let us know, what do you think or share with us your knowledge: