Redirect on an exception in Flask using a decorator

In a Flask application, you typically implement global exception handlers. In many cases, this is sufficient. But what if you want more control?

In one project, I was connecting to an API and I wanted a number of routes that used the API to redirect to a 'start' page in case of an API error, with an appropriate message of course. I implemented this using a 'redirect_decorator' exception handler that also has a parameter specifying the endpoint. The decorator is used to avoid having to give each route try-except code.

Project setup

As always, create a virtual environment, and directories, for example by typing:

Translated with www.DeepL.com/Translator (free version)
python3 -m venv flask_except
source flask_except/bin/activate
cd flask_except
mkdir project
cd project
mkdir app
pip install flask

In the project directory there is the file to run the application:

# run.py
from app.factory import create_app

app = create_app()

if __name__ == '__main__':
    app.run(
        host='localhost',
        debug=True,
    )

Flask with an error handler

This is more or less a copy of what is on the Flask website.

# factory.py
from flask import Flask

def internal_server_error(e):
    return 'An internal server error occurred', 500

def create_app():
    
    app = Flask(__name__)
    
    @app.route('/')
    def welcome():
        return 'Welcome'

    @app.route('/error')
    def error():
        a = 2/0
        return 'Error'

    app.register_error_handler(500, internal_server_error)

    return app

In the project directory we start the application:

python run.py

Then in the browser you can check the endpoints:

http://127.0.0.1:5000
http://127.0.0.1:5000/error

If you want to see the internal_server_error() function message then set debug=False in run.py. In this case the server is not restarted on changes!

Add an API that generates exceptions on errors

In Python, most external packages generate exceptions when errors occur. That is what we are going to simulate here. Our API package has one function and generates its own exceptions.

# api: custom exception
class ServiceAPIError(Exception):
    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs

# api: function
def service_api_function():
    if True:
        raise ServiceAPIError(
            reason='Connection problem',
            description='Remote system not responding',
        )

If an error occurs, the function raises an ServiceAPIError with a reason and description.

The redirect decorator

For maximum flexibility I want to be able to pass an endpoint in the decorator. This means wrapping the decorator again.

# our redirect decorator
def redirect_on_service_api_error(endpoint):
    def decorator(f):
        @functools.wraps(f)
        def func(*args, **kwargs):
            try:
                return f(*args, **kwargs)
            except ServiceAPIError as e:
                return redirect(url_for(endpoint))
        return func
    return decorator

Adding the API and redirect decorator

Our application has one or more routes to execute tasks. When a task fails because of an API error,  we want to be redirected to the route 'start'. I added print statements to see all information available in our decorator:

# factory.py (with api, redirect decorator)
from flask import Flask, redirect, url_for
import functools

# api: custom exception
class ServiceAPIError(Exception):
    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs

# api: function
def service_api_function():
    if True:
        raise ServiceAPIError(
            reason='Connection problem',
            description='Remote system not responding',
        )

# our redirect decorator
def redirect_on_service_api_error(endpoint):
    def decorator(f):
        @functools.wraps(f)
        def func(*args, **kwargs):
            try:
                return f(*args, **kwargs)
            except ServiceAPIError as e:
                # debugging
                print('Exception: {}, e = {}'.format(type(e).__name__, e))
                print('- function = {}, args = {}, kwargs = {}'.format(f.__name__, args, kwargs))
                print('- endpoint = {}'.format(endpoint))
                print('- e.args = {}, e.kwargs = {}'.format(e.args, e.kwargs))
                # redirect
                return redirect(url_for(endpoint))
        return func
    return decorator

def internal_server_error(e):
    return 'An internal server error occurred', 500

def create_app():
    
    app = Flask(__name__)

    @app.route('/')
    def welcome():
        return 'Welcome'

    @app.route('/error')
    def error():
        a = 2/0
        return 'Error'

    @app.route('/start')
    def start():
        return 'Start'

    @app.route('/tasks/<task>')
    @redirect_on_service_api_error('start')
    def tasks(task):
        print('tasks: task = {}'.format(task))
        service_api_function()
        return 'Started task'

    app.register_error_handler(500, internal_server_error)

    return app

Now point your browser to:

http://127.0.0.1:5000/tasks/mytask

and observe that we are redirected to 'start'. Nice. In the console the following is printed:

Exception: ServiceAPIError, e = 
- function = tasks, args = (), kwargs = {'task': 'mytask'}
- endpoint = start
- e.args = (), e.kwargs = {'reason': 'Connection problem', 'description': 'Remote system not responding'}

But we can do better: Message Flashing

When there is an API error and we are redirected to 'start', we also want to show what happened. We can do this using the flash() message function in Flask. Before redirecting in our decorator, we flash() the message:

flash("There was a problem in '{}' with task  '{}': {}".\
    format(f.__name__, kwargs['task'], e.kwargs['reason']))

I will not implement this in templates but instead use the get_flashed_messages() function in the 'start' route:

    @app.route('/start')
    def start():
        return 'Start<br>' + ', '.join(get_flashed_messages())

To use flash() we also must add the secret_key to our app.

Our final application:

# factory.py (with api, redirect decorator, flashed messages)
from flask import Flask, redirect, url_for, flash, get_flashed_messages
import functools

# api: custom exception
class ServiceAPIError(Exception):
    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs

# api: function
def service_api_function():
    if True:
        raise ServiceAPIError(
            reason='Connection problem',
            description='Remote system not responding',
        )

# our redirect decorator
def redirect_on_service_api_error(endpoint):
    def decorator(f):
        @functools.wraps(f)
        def func(*args, **kwargs):
            try:
                return f(*args, **kwargs)
            except ServiceAPIError as e:
                # debugging
                print('Exception: {}, e = {}'.format(type(e).__name__, e))
                print('- function = {}, args = {}, kwargs = {}'.format(f.__name__, args, kwargs))
                print('- endpoint = {}'.format(endpoint))
                print('- e.args = {}, e.kwargs = {}'.format(e.args, e.kwargs))
                # flash
                flash("There was a problem in '{}' with task  '{}': {}".\
                    format(f.__name__, kwargs['task'], e.kwargs['reason']))
                # redirect
                return redirect(url_for(endpoint))
        return func
    return decorator

def internal_server_error(e):
    return 'An internal server error occurred', 500

def create_app():
    
    app = Flask(__name__)

    app.secret_key = b'_5#y2L"F4Q8z\n\xec]/'

    @app.route('/')
    def welcome():
        return 'Welcome'

    @app.route('/error')
    def error():
        a = 2/0
        return 'Error'

    @app.route('/start')
    def start():
        return 'Start<br>' + ', '.join(get_flashed_messages())

    @app.route('/tasks/<task>')
    @redirect_on_service_api_error('start')
    def tasks(task):
        print('tasks, task = {}'.format(task))
        service_api_function()
        return 'Started task'

    app.register_error_handler(500, internal_server_error)

    return app

Now point your browser to:

http://127.0.0.1:5000/tasks/mytask

and we are redirected to 'start', but also the message is displayed. Now we know what happened!

Start
There was a problem in 'tasks' with task 'mytask': Connection problem

Summary

Compared to try-except in every route we have much less code. And in our redirect decorator exception handler we have all the information we need to show a decent message.

Links / credits

Flask - Handling Application Errors
https://flask.palletsprojects.com/en/2.1.x/errorhandling

Flask - Message Flashing
https://flask.palletsprojects.com/en/2.1.x/patterns/flashing

How do I pass extra arguments to a Python decorator?
https://stackoverflow.com/questions/10176226/how-do-i-pass-extra-arguments-to-a-python-decorator