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

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

Нет.

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

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

Мне пришло в голову, что это просто реализовать стандартную транзакцию базы данных диспетчер контекста, но разрешить ему принимать 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')

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

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

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

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

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

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

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 предлагает декораторам сделать это, так что вам не нужно кататься самостоятельно. Существует даже диспетчер контекста, который эффективно действует как временный декоратор для вашей функции, но он по-прежнему полагается на извлекаемый код, который будет разложен на единую функцию.

Этому вопросу несколько лет, но после прочтения ответов я решил попробовать.

Это решение требует использования вспомогательного класса, но я думаю, что оно предоставляет интерфейс с повторными попытками, настроенными через диспетчер контекста.

class Client:
    def _request(self):
        # do request stuff
        print("tried")
        raise Exception()

    def request(self):
        retry = getattr(self, "_retry", None)
        if not retry:
            return self._request()
        else:
            for n in range(retry.tries):
                try:
                    return self._request()
                except Exception:
                    retry.attempts += 1


class Retry:
    def __init__(self, client, tries=1):
        self.client = client
        self.tries = tries
        self.attempts = 0

    def __enter__(self):
        self.client._retry = self

    def __exit__(self, *exc):
        print(f"Tried {self.attempts} times")
        del self.client._retry

>>> client = Client()
>>> with Retry(client, tries=3):
    ... # will try 3 times
    ... response = client.request()

tried once
tried once
tried once
Tried 3 times