Обработка ошибок в стиле panic/defer на Python

Обработка ошибок в Go построена не на закостенелом механизме исключений, а на новом интересном механизме отложенных обработчиков. В качестве интересного исследования я реализовал такую обработку ошибок на Python. Кому интересно, заходите.

Принцип работы обработки ошибок в Go следующий, вы указываете ключевое слово defer, после которого ставите вызов функции, который выполнится при завершении метода: обычном или паническом (при возникновении ошибки). Пример:

func CopyFile(dstName, srcName string) (written int64, err error) {
    src, err := os.Open(srcName)
    if err != nil {
        return
    }
    defer src.Close()

    dst, err := os.Create(dstName)
    if err != nil {
        return
    }
    defer dst.Close()

    return io.Copy(dst, src)
}

Подробнее можете почитать здесь. При указании отложенной функции фиксируются аргументы, а вызов происходит в конце содержащей их функции. Если вы хотите прервать выполнение функции с состоянием ошибки, необходимо вызвать функцию panic(). При этом в порядке, обратном установке, вызываются отложенные функции. Если в одной из них вызывается функция recover(), то ошибочное состояние снимается, и после возврата из метода выполнение программы пойдёт в привычном порядке.

Подобное поведение можно реализовать на Python, благодаря гибкости языка. Для этого объявляются соответствующие функции, которые используют специальные переменные в стеке, чтобы вешать обработчики на функцию, и устанавливать специальный статус в случае восстановления. Для указания в функции поддержки данного механизма используется декоратор, который создаём список для хранения отложенных функций, и перехватывает исключение для их вызова. Код:

# Go-style error handling

import inspect
import sys

def panic(x):
    raise Exception(x)

def defer(x):
    for f in inspect.stack():
        if '__defers__' in f[0].f_locals:
            f[0].f_locals['__defers__'].append(x)
            break

def recover():
    val = None
    for f in inspect.stack():
        loc = f[0].f_locals
        if f[3] == '__exit__' and '__suppress__' in loc:
            val = loc['exc_value']
            loc['__suppress__'].append(True)
            break
    return val

class DefersContainer(object):
    def __init__(self):
        # List for sustain refer in shallow clone
        self.defers = []

    def append(self, defer):
        self.defers.append(defer)

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_value, traceback):
        __suppress__ = []
        for d in reversed(self.defers):
            try:
                d()
            except:
                __suppress__ = []
                exc_type, exc_value, traceback = sys.exc_info()
        return __suppress__


def defers_collector(func):
    def __wrap__(*args, **kwargs):
        __defers__ = DefersContainer()
        with __defers__:
            func(*args, **kwargs)
    return __wrap__


@defers_collector
def func():
    f = open('file.txt', 'w')
    defer(lambda: f.close())

    defer(lambda : print("Defer called!"))

    def my_defer():
        recover()

    defer(lambda: my_defer())

    print("Ok )")
    panic("WTF?")

    print("Never printed (((")


func()
print("Recovered!")

Я использую lambda для фиксации аргументов при отложенном вызове, чтобы повторить поведение оператора defer.

Функциональную идентичность в нюансах не тестировал. Но если знаете что нужно доработать, пишите.

Автор: deko

Источник