Инкапсуляция попыток в блок `with`

Ответ 1

Можно ли повторить код в инструкции with?

Нет

Как указывалось ранее в этом потоке списка рассылки, вы можете уменьшить количество дубликатов, заставив декодера вызвать переданную функцию:

def do_work():
    ...
    # This is not ideal!
    @transaction(retries=3)
    def _perform_in_transaction():
        # Atomic DB statements
        ...
    # called implicitly
    ...

Ответ 2

Мне кажется, что для этого нужно просто реализовать стандартный менеджер контекста транзакции базы данных, но позволить ему принять аргумент retries в конструкторе. Тогда я бы просто обернул это в ваших реализациях методов. Что-то вроде этого:

class transaction(object):
    def __init__(self, retries=0):
        self.retries = retries
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, traceback):
        pass

    # Implementation...
    def execute(self, query):
        err = None
        for _ in range(self.retries):
            try:
                return self._cursor.execute(query)
            except Exception as e:
                err = e # probably ought to save all errors, but hey
        raise err

with transaction(retries=3) as cursor:
    cursor.execute('BLAH')

Ответ 3

Поскольку декораторы - это просто функции, вы можете сделать следующее:

with transaction(_perform_in_transaction, retries=3) as _perf:
    _perf()

Для получения дополнительной информации вам нужно реализовать transaction() как метод factory, который возвращает объект с __callable__(), установленным для вызова исходного метода, и повторите его до retries количество раз при ошибке; __enter__() и __exit__() будут определены как обычные для контекстных менеджеров транзакций базы данных.

Вы можете альтернативно настроить transaction() таким образом, чтобы он сам выполнял переданный метод до retries количество раз, что, вероятно, потребует примерно такой же работы, как реализация диспетчера контекста, но будет означать, что фактическое использование будет уменьшен до просто transaction(_perform_in_transaction, retries=3) (что фактически эквивалентно приложению декоратора delnan).

Ответ 4

Хотя я согласен, что это не может быть сделано с помощью менеджера контекста... это может быть сделано с двумя менеджерами контекста!

Результат немного неловкий, и я не уверен, одобряю ли я свой собственный код, но вот как он выглядит как клиент:

with RetryManager(retries=3) as rm:
    while rm:
        with rm.protect:
            print("Attempt #%d of %d" % (rm.attempt_count, rm.max_retries))
             # Atomic DB statements

Существует явный, while цикл до сих пор, и не один, а два, with заявлениями, что оставляет слишком много возможностей для ошибок по моей душе.

Вот код:

class RetryManager(object):
    """ Context manager that counts attempts to run statements without
        exceptions being raised.
        - returns True when there should be more attempts
    """

    class _RetryProtector(object):
        """ Context manager that only raises exceptions if its parent
            RetryManager has given up."""
        def __init__(self, retry_manager):
            self._retry_manager = retry_manager

        def __enter__(self):
            self._retry_manager._note_try()
            return self

        def __exit__(self, exc_type, exc_val, traceback):
            if exc_type is None:
                self._retry_manager._note_success()
            else:
                # This would be a good place to implement sleep between
                # retries.
                pass

            # Suppress exception if the retry manager is still alive.
            return self._retry_manager.is_still_trying()

    def __init__(self, retries=1):

        self.max_retries = retries
        self.attempt_count = 0 # Note: 1-based.
        self._success = False

        self.protect = RetryManager._RetryProtector(self)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, traceback):
        pass

    def _note_try(self):
        self.attempt_count += 1

    def _note_success(self):
        self._success = True

    def is_still_trying(self):
        return not self._success and self.attempt_count < self.max_retries

    def __bool__(self):
        return self.is_still_trying()

Бонус: я знаю, что вы не хотите разделять свою работу на отдельные функции, обернутые декораторами... но если вы были довольны этим, пакет redo от Mozilla предлагает декораторам сделать это, так что вам не нужно катись самостоятельно. Существует даже диспетчер контекста, который эффективно действует как временный декоратор для вашей функции, но он все еще полагается на ваш извлекаемый код, который будет выделен в одну функцию.

Ответ 5

Вместо использования менеджера контекста вы можете использовать генератор для элегантного достижения своей цели (без встроенной функции).

Используйте модуль повтора, сайт вызова будет выглядеть так:

for _ in redo.retrier(attempts=3, sleeptime=1):
    try:
        with transaction:
            # Atomic DB statements
        if some reasons need to retry:
            continue
        break  # it done
    except ExceptionsToRetry:
        continue
    # For exceptions not caught, just bail out
else:
    print('max retry hint')