Перенаправление на исключение в Flask с помощью decorator

В приложении Flask обычно реализуются глобальные обработчики исключений. Во многих случаях этого достаточно. Но что, если вам нужно больше контроля?

В одном проекте я подключался к API и хотел, чтобы несколько маршрутов, использующих API , перенаправляли на "стартовую" страницу в случае ошибки API , с соответствующим сообщением, конечно. Я реализовал это с помощью обработчика исключений 'redirect_decorator', который также имеет параметр, указывающий конечную точку. decorator используется для того, чтобы избежать необходимости передавать код try-except каждому маршруту.

Настройка проекта

Как обычно, создайте virtual environment, и каталоги, например, набрав:

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

В директории проекта находится файл для запуска приложения:

# run.py
from app.factory import create_app

app = create_app()

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

Flask с обработчиком ошибок.

Это более или менее копия того, что находится на сайте Flask .

# 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

В директории проекта запускаем приложение:

python run.py

Затем в браузере вы можете проверить конечные точки:

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

Если вы хотите увидеть сообщение функции internal_server_error(), то установите debug=False в run.py. В этом случае сервер не перезапускается при изменениях!

Добавьте API , который генерирует исключения при ошибках

В Python большинство внешних пакетов генерируют исключения при возникновении ошибок. Именно это мы и собираемся смоделировать здесь. Наш пакет API имеет одну функцию и генерирует свои собственные исключения.

# 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',
        )

Если возникает ошибка, функция поднимает ServiceAPIError с причиной и описанием.

Перенаправление decorator

Для максимальной гибкости я хочу иметь возможность передавать конечную точку в decorator. Это означает, что нужно снова обернуть decorator .

# 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

Добавление API и перенаправление decorator

Наше приложение имеет один или несколько маршрутов для выполнения задач. Когда задача не выполняется из-за ошибки API , мы хотим быть перенаправлены на маршрут 'start'. Я добавил операторы печати, чтобы увидеть всю информацию, имеющуюся в нашем 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

Теперь направьте браузер на:

http://127.0.0.1:5000/tasks/mytask

и увидите, что мы перенаправлены на 'start'. Отлично. В консоли выводится следующее:

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

Но мы можем сделать лучше: Мигающее сообщение

Когда возникает ошибка API и мы перенаправляемся на 'start', мы также хотим показать, что произошло. Мы можем сделать это с помощью функции flash() сообщения в Flask. Перед перенаправлением в нашем decorator мы показываем flash() сообщение:

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

Я не буду реализовывать это в шаблонах, а вместо этого использую функцию get_flashed_messages() в маршруте 'start':

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

Чтобы использовать flash(), мы также должны добавить ключ secret_key в наше приложение.

Наше конечное приложение:

# 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(task):
        print('tasks, task = {}'.format(task))
        service_api_function()
        return 'Started task'

    app.register_error_handler(500, internal_server_error)

    return app

Теперь направьте свой браузер на:

http://127.0.0.1:5000/tasks/mytask

и мы будем перенаправлены на 'start', но также будет показано сообщение. Теперь мы знаем, что произошло!

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

Резюме

По сравнению с try-except в каждом маршруте у нас гораздо меньше кода. А в нашем обработчике исключений redirect decorator у нас есть вся информация, необходимая для вывода приличного сообщения.

Ссылки / кредиты

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