У меня есть кусок кода в Python, который, вероятно, вызывает ошибку вероятностно, потому что он обращается к серверу, и иногда на этом сервере имеется ошибка внутреннего внутреннего сервера. Я хочу продолжать попытки, пока не получу ошибку. Мое решение было:
while True:
try:
#code with possible error
except:
continue
else:
#the rest of the code
break
Это кажется мне взломанным. Есть ли более питонический способ сделать это?
Это не станет намного чище. Это не очень чистая вещь. В лучшем случае (что в любом случае было бы более читабельным, поскольку условие break остается там с while), вы можете создать переменную result = None и выполнить цикл, пока он is None. Вы также должны скорректировать переменные, и вы можете заменить continue семантически, возможно, правильным pass (вам все равно, если произойдет ошибка, вы просто хотите игнорировать ее) и отбросить break - это также возвращает остаток кода, который выполняется только один раз, вне цикла. Также обратите внимание, что голые except: пункты являются злыми по причинам, указанным в документации.
Пример, включающий все вышеперечисленное:
result = None
while result is None:
try:
# connect
result = get_data(...)
except:
pass
# other code that uses result but is not involved in getting it
Возможно, что-то вроде этого:
connected = False
while not connected:
try:
try_connect()
connected = True
except ...:
pass
Вот один, который тяжело терпит неудачу после 4 попыток и ждет 2 секунды между попытками. Измените, как вы хотите получить то, что хотите, из этого:
from time import sleep
for x in range(0, 4): # try 4 times
try:
# msg.send()
# put your logic here
str_error = None
except Exception as str_error:
pass
if str_error:
sleep(2) # wait for 2 seconds before trying to fetch the data again
else:
break
Вот пример с отсрочкой:
from time import sleep
sleep_time = 2
num_retries = 4
for x in range(0, num_retries):
try:
# put your logic here
str_error = None
except Exception as str_error:
pass
if str_error:
sleep(sleep_time) # wait before trying to fetch the data again
sleep_time *= 2 # Implement your backoff algorithm here i.e. exponential backoff
else:
break
Здесь функция утилиты, которую я написал, чтобы обернуть повторную попытку, пока успех не достигнет более аккуратного пакета. Он использует ту же основную структуру, но предотвращает повторение. Его можно было бы модифицировать, чтобы поймать и повторить исключение в финальной попытке относительно легко.
def try_until(func, max_tries, sleep_time):
for _ in range(0,max_tries):
try:
return func()
except:
sleep(sleep_time)
raise WellNamedException()
#could be 'return sensibleDefaultValue'
Затем можно называть это
result = try_until(my_function, 100, 1000)
Если вам нужно передать аргументы в my_function, вы можете сделать это, указав try_until пересылку аргументов или обернув его без аргумента lambda:
result = try_until(lambda : my_function(x,y,z), 100, 1000)
Рецепты itertools.iter_except инкапсулируют эту идею "повторного вызова функции до возникновения исключения". Это похоже на принятый ответ, но рецепт дает вместо этого итератор.
Из рецептов:
def iter_except(func, exception, first=None):
""" Call a function repeatedly until an exception is raised."""
try:
if first is not None:
yield first() # For database APIs needing an initial cast to db.first()
while True:
yield func()
except exception:
pass
Вы, конечно, можете реализовать последний код напрямую. Для удобства я использую отдельную библиотеку more_itertools, которая реализует этот рецепт для нас (необязательно).
Код
import more_itertools as mit
list(mit.iter_except([0, 1, 2].pop, IndexError))
# [2, 1, 0]
подробности
Здесь метод pop (или заданная функция) вызывается для каждой итерации объекта списка до тех пор, пока не будет IndexError.
Для вашего случая, учитывая некоторую connect_function и ожидаемую ошибку, вы можете создать итератор, который будет вызывать функцию несколько раз, пока не возникнет исключение, например
mit.iter_except(connect_function, ConnectionError)
На этом этапе обработайте его как любой другой итератор, зацикливая его или вызывая next().
Может быть, на основе декоратора? Вы можете передать в качестве аргументов декоратора список исключений, по которым мы хотим повторить попытку и/или количество попыток.
def retry(exceptions=None, tries=None):
if exceptions:
exceptions = tuple(exceptions)
def wrapper(fun):
def retry_calls(*args, **kwargs):
if tries:
for _ in xrange(tries):
try:
fun(*args, **kwargs)
except exceptions:
pass
else:
break
else:
while True:
try:
fun(*args, **kwargs)
except exceptions:
pass
else:
break
return retry_calls
return wrapper
from random import randint
@retry([NameError, ValueError])
def foo():
if randint(0, 1):
raise NameError('FAIL!')
print 'Success'
@retry([ValueError], 2)
def bar():
if randint(0, 1):
raise ValueError('FAIL!')
print 'Success'
@retry([ValueError], 2)
def baz():
while True:
raise ValueError('FAIL!')
foo()
bar()
baz()
конечно, часть "try" должна быть перенесена в другую функцию, потому что мы используем ее в обеих циклах, но это просто пример;)
Как и большинство других, я бы рекомендовал пробовать конечное количество раз и спать между попытками. Таким образом, вы не окажетесь в бесконечном цикле на случай, если что-то действительно случится с удаленным сервером.
Я также рекомендую продолжать только тогда, когда вы получите конкретное исключение, которое вы ожидаете. Таким образом, вы все равно можете обрабатывать исключения, которые вы можете не ожидать.
from urllib.error import HTTPError
import traceback
from time import sleep
attempts = 10
while attempts > 0:
try:
#code with possible error
except HTTPError:
attempts -= 1
sleep(1)
continue
except:
print(traceback.format_exc())
#the rest of the code
break
Кроме того, вам не нужен блок еще. Из-за продолжения в блоке исключений вы пропускаете оставшуюся часть цикла, пока блок попытки не сработает, пока не будет выполнено условие while или не возникнет исключение, отличное от HTTPError.
e = ''
while e == '':
try:
response = ur.urlopen('https://https://raw.githubusercontent.com/MrMe42/Joe-Bot-Home-Assistant/mac/Joe.py')
e = ' '
except:
print('Connection refused. Retrying...')
time.sleep(1)
Это должно работать. Он устанавливает e в '', и цикл while проверяет, является ли он все еще ''. Если обнаружена ошибка в операторе try, он печатает, что в соединении было отказано, ждет 1 секунду и затем начинается заново. Это будет продолжаться до тех пор, пока не будет ошибки в try, который затем устанавливает e в '', что убивает цикл while.
Вот короткая часть кода, которую я использую для захвата ошибки в виде строки. Будет повторять, пока это не удастся. Это улавливает все исключения, но вы можете изменить это, как хотите.
start = 0
str_error = "Not executed yet."
while str_error:
try:
# replace line below with your logic , i.e. time out, max attempts
start = raw_input("enter a number, 0 for fail, last was {0}: ".format(start))
new_val = 5/int(start)
str_error=None
except Exception as str_error:
pass
ПРЕДУПРЕЖДЕНИЕ: Этот код будет зацикливаться на бесконечном цикле до тех пор, пока не произойдет исключение. Это просто простой пример, и MIGHT требует, чтобы вы быстрее вышли из цикла или поспали между повторами.
При повторной попытке из-за ошибки вы всегда должны:
Простой общий способ решения этой проблемы при рассмотрении этих проблем заключается в использовании библиотеки отката. Основной пример:
import backoff
@backoff.on_exception(
backoff.expo,
MyException,
max_tries=5
)
def make_request(self, data):
# do the request
Этот код оборачивает make_request декоратором, который реализует логику повторных попыток. Мы повторяем каждый раз, когда происходит наша конкретная ошибка MyException, с ограничением в 5 попыток. Экспоненциальный откат является хорошей идеей в этом контексте, чтобы помочь минимизировать дополнительную нагрузку, которую наши повторные попытки возлагают на удаленный сервер.
как насчет повторной библиотеки на pypi? Я использую его некоторое время, и он делает именно то, что мне нужно, и даже больше (повторите при ошибке, повторите, когда нет, повторите с таймаутом). Ниже приведен пример с их сайта:
import random
from retrying import retry
@retry
def do_something_unreliable():
if random.randint(0, 10) > 1:
raise IOError("Broken sauce, everything is hosed!!!111one")
else:
return "Awesome sauce!"
print do_something_unreliable()