Best way to handle repeated, complex exception handling?

Viewed 59

I'm looking to refactor some code to make it more concise, and I'm a little stumped on one aspect. There's a good chance I may be missing something obvious.

The application is a webserver, where each API endpoint is a separate method of a RequestHandler class (Tornado specifically, but I'd like to know a general approach). So essentially, something like:

class MyHandler(tornado.web.RequestHandler):
    def endpoint_1(self, user_id):
        ...
 
    def endpoint_2(self, user_id):
        ...

Many of the endpoints need to lookup a user via the database, so I added a method like def get_user(self, user_id) to the class. This method also takes care of various permission checks required of the user, and as such there are several ways it can fail, and each failure should send an appropriate message back to the user and close the connection.

My question then is how best to call this method within each endpoint method in a way that is both concise and obvious.

I see three possible approaches:

  1. Catch all relevant exceptions within each endpoint method:
def endpoint_1(self, user_id):
   try:
       self.get_user(user_id)
   except UserDoesNotExist:
       self.write("No such user")
       self.connection.close()
       return
   except MissingPermission1:
       self.write("No permission 1")
       self.connection.close()
       return
   ...

This to me is the most clear as to control flow, but requires making sure these except: blocks are properly copied into each method.

  1. Unwrapped get_user() call, let get_user() handle closing the connection
def get_user(self, user_id):
    try:
        user = self.database.lookup(user_id)
        if not user.hasPermission(1):
            raise MissingPermission1
        ...
    except UserDoesNotExist:
        self.write("No such user")
        self.connection.close()
    except MissingPermission1:
        self.write("No permission 1")
        self.connection.close()
    ...

def endpoint_1(self, user_id):
    user = self.get_user(user_id)
    ...

This option is by far the most concise but something feels "dirty" to me about letting get_user() close the connection, preemptively ending the control flow of endpoint_1(). Maybe this is just me though? Documentation could solve this issue, but I still just get bad code smells from this.

  1. Catch-all except, pass to an exception-handling method
def get_user(self, user_id):
    user = self.database.lookup(user_id)
    if not user.hasPermission(1):
        raise MissingPermission1
    ...

def handle_exception(self, exc):
    if isinstance(exc, MissingPermission1):
        self.write("Missing permission 1")
        return
    ...

def endpoint_1(self, user_id):
    try:
        self.get_user(user_id)
    except Exception as e:
        self.handle_exception(e)
        self.connection.close()
        return

This is the approach I like the most, but something about it still feels off.

I'm very curious as to the what others think on this, preferably in as general as possible (e.g. not Tornado/webserver specific).

2 Answers

I'll introduce another approach that has worked for me when working with Desktop GUIs.

The idea is similar to your method 3, but instead of having to add a try/except to each endpoint... method, we simply override sys.excepthook with our own "custom excepthook method". This means that your endpoint... methods are now free to let the exception be raised all the way to the custom excepthook.

This is a minimal working example:

import random
import sys

class UserDoesNotExist(Exception): pass
class MissingPermission1(Exception): pass

class MyApp:
    def __init__(self):
        sys.excepthook = self.app_excepthook
        
    def app_excepthook(self, exc_type, exc_value, exc_traceback):
        if issubclass(exc_type, UserDoesNotExist):
            self.handle_inexistent_user()
        elif issubclass(exc_type, MissingPermission1):
            self.handle_missing_permission()
        else:
            sys.__excepthook__(exc_type, exc_value, exc_traceback)

    def handle_inexistent_user(self):
        self.write("No such user")
        # self.connection.close(), etc...

    def handle_missing_permission(self):
        self.write("No permission 1")
        # self.connection.close(), etc...
    
    @staticmethod
    def write(message):
        print(message)
    
    def try_to_connect(self):
        """Simulates a connection that works 25% of the time"""
        r = random.random()
        if r < 0.25:
            raise UserDoesNotExist
        elif r < 0.50:
            raise MissingPermission1
        elif r < 0.75:
            raise ValueError
        else:
            print('Success')

if __name__ == '__main__':
    app = MyApp()
    app.try_to_connect()

Like I said, the idea here is to override sys.excepthook with the app_excepthook method. This method checks for the exception type and calls the appropriate handler method. If no method has been delegated for a particular exception type, sys.__excepthook__ deals with it. This is simply a pointer to the original sys.excepthook, which means that any exception not explicitly dealt by your App will be dealt by Python's default excepthook.

If you try executing this a couple times, you'll see that UserDoesNotExist and MissingPermission1 exceptions are properly dealt with by your App, while the ValueError goes all the way up to the default excepthook.

More importantly: the try_to_connect method does not have to worry about any try/except clauses, or calling a specific method depending on the type of exception. It simply raises the exception and lets the App itself deal with it.

You could even abstract this further and move all exception handling functionality into a dedicated ExceptionHandler class that becomes an instance attribute of MyApp upon initialization.

Some considerations/drawbacks of this approach:

  • It may not be simple to give additional information to the handle_... methods as parameters, since app_excepthook has to follow the signature of sys.excepthook. This means that everything that the handle... methods need must be stored as instance attributes, i.e. self.connection, self.thing, etc. You can work around this by modifying your custom exceptions to accept additional args / kwargs, although I'm not sure how "hacky" that would look.
  • MyApp is expected to be a singleton, since multiple instances of the class will move the sys.excepthook pointer to the latest instance created.
  • Likewise, any other sort of interactions with sys.excepthook will likely result in undesired behavior. For example, this code does not work as expected in an IPython shell because it also messes with sys.excepthook on startup.

Context manager (via e.g. contextlib.contextmanager) seems applicable to your use case since it will allow to you to:

  • Get the user and handle the exceptions in 1 place (no need to repeat per method that needs it). Thus no need of manual try-except blocks scattered around
  • Close a resource once the processing has been completed for user.
from contextlib import contextmanager


class MyHandler():
    def __init__(self, database):
        self.database = database

    @contextmanager
    def get_user(self, user_id):
        user = None
        try:
            user = self.database[user_id]
        except KeyError:
            print(f"Missing user {user_id}")

        try:
            yield user
        finally:
            print(f"Closing resources for {user_id}")
            self.connection = "closed"

    def endpoint_1(self, user_id):
        self.connection = "open"
        with self.get_user(user_id) as user:
            if user is None:
                # Handle scenario if user wasn't successfully retrieved
                print(f"Can't process {user_id}")
            else:
                # Handle scenario if user was successfully retrieved
                print(f"Successfully retrieved {user}")


handler = MyHandler({"a1": "90s Band", "b2": "Banana", "c3": "Bomb"})
print("===== Call 1 =====")
handler.endpoint_1("a1")
print("===== Call 2 =====")
handler.endpoint_1("b1")
print("===== Call 3 =====")
handler.endpoint_1("b2")
print("===== Call 4 =====")
handler.endpoint_1("c3")

Output

===== Call 1 =====
Successfully retrieved 90s Band
Closing resources for a1
===== Call 2 =====
Missing user b1
Can't process b1
Closing resources for b1
===== Call 3 =====
Successfully retrieved Banana
Closing resources for b2
===== Call 4 =====
Successfully retrieved Bomb
Closing resources for c3
  • All resources will only be closed upon exit of the with statement. Thus you can perform code safely within the with statement knowing that the resources are not yet closed. Also, no need to remember to close the resources since it is automatically performed upon exit of the with statement.

Another alternative you might explore is using decorators to wrap around your methods.

Related