Python-кода с функцией повтора


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

Я написал следующую функцию, но думаю, если есть лучший способ сделать это?

Пример функции я хочу повторить для

def is_user_exist(username):

    try:
        pwd.getpwnam(username)
        log.info("User %s exist" % username)
        return True
    except KeyError:
        log.exception("User %s does not exist." % username)
        return False

Моя функция повтора

def retry(func, *func_args, **kwargs):
    retry_count = kwargs.get("retry_count", 5)
    delay = kwargs.get("delay", 5)
    while retry_count > 1:
        if func(*func_args):
            return
        log.debug("waiting for %s seconds before retyring again")
        sleep(delay)
        retry_count = retry_count - 1

    return func(*func_args)


7187
17
задан 28 февраля 2018 в 03:02 Источник Поделиться
Комментарии
5 ответов

Я как и все ев. Kounis' ответа, поэтому я добавлю выше уровня деталей.

Пусть это будет правда-г

Сейчас вы не строго требуя func вернуться True или Falseпросто что-то правда или ложь-г. Это нормально, и обновления. Я думаю, что вы выиграли бы от указывая, что в строкой документации (который вы также должны писать). Кроме того, может быть, стоит вернуть фактическое значение func(*func_args) вместо True или False; таким образом, вы можете на самом деле получить значение из функции, если вы хотели воспользоваться истины-yness стоимости.

Лучше kwargs поддержка

Вы также не допустить каких-либо аргументов сайта, чтобы быть переданный функции - функции, может поддерживать их, так что вы должны получить. Я бы продвигать два сайта Аргументы, вытащить из kwargs к явным ключевое слово аргументы, с подходящими значениями по умолчанию (5 в вашем случае).

Исключения

Мне странно, что эта функция не имеет никакого понятия о повтором после исключения. Я не хочу, чтобы ты делал что-то подобное, на что я надеюсь, понятным причинам

for _ in range(retry_count):
try:
if func(*func_args):
return True
except:
pass
log.debug("wating for %s seconds before retrying again")
sleep delay)

Однако, во многих случаях, подозреваю, вы бы знали, какие исключения вы могли бы ожидать (например, сеть тайм-аут или подключение вспышка) и что вы, возможно, хотите быть обработан повтор рамках. С этой целью, я думаю, что это может быть приятно:

def retry(func, *args, retry_count=5, delay=5, allowed_exceptions=(), **kwargs):
for _ in range(retry_count):
try:
result = func(*args, **kwargs)
if result: return result
except allowed_exceptions as e:
pass

Это, очевидно, не полная реализация; я оставил некоторые другие предметы у вас есть, и он ведет себя странно, если он не на последней итерации, но это должно быть достаточно, чтобы начать.

Модные вещи

Я думаю, мы также могли бы получить больше пользы от этого, если он был декоратором. Тогда потребители функция даже не нужно знать, если они хотят это повторить или нет; они просто называют свою функцию и вижу, что это работает, и действительно ли он был повторен становится неактуальным. Не забывайте использовать functools.wraps для сохранения метаданных.

import functools

def retry(retry_count=5, delay=5, allowed_exceptions=()):
def decorator(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
for _ in range(retry_count):
# everything in here would be the same

return wrapper
return decorator

Затем вы можете включить повтор для всех, вот так:

@retry(retry_count=5, delay=5)
def is_user_exist(username):
try:
pwd.getpwnam(username)
log.info("User %s exist" % username)
return True
except KeyError:
log.exception("User %s does not exist." % username)
return False

Действительно модные вещи

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

По awaitИнг асинхронная функция, которая просто работает n секунд, вы достигаете той же цели, но позволяет другим асинхронные функции, чтобы сделать работу, пока вы просто ждете. Обратите внимание, что в зависимости от цикла обработки событий вы можете в конечном итоге ждет немного больше или меньше времени.

Я также очищены вопросов я упомянул об обработке исключений; оно теперь всегда возвращает результат функции, если он имеет один, и он будет ре-рейз исключение без потери любой трассировки, если там был один. Что также использует питон только 3 функции; я оставил комментарий, как это сделать в Python 2.

Обратите внимание, я не так хорошо знаком с ввода-вывода, Как я никогда не должен делать какие-либо серьезные разработки есть, так что я не мог иметь этот кусок кода точно; теория должна быть хоть звук.

import functools
import asyncio

def retry(retry_count=5, delay=5, allowed_exceptions=()):
def decorator(f):
@functools.wraps(f)
async def wrapper(*args, **kwargs):
result = None
last_exception = None
for _ in range(retry_count):
try:
result = func(*func_args, **kwargs)
if result: return result
except allowed_exceptions as e:
last_exception = e
log.debug("Waiting for %s seconds before retrying again")
await asyncio.sleep(delay)

if last_exception is not None:
raise type(last_exception) from last_exception
# Python 2
# import sys
# raise type(last_exception), type(last_exception)(last_exception), sys.exc_info()[2]

return result

return wrapper
return decorator

16
ответ дан 28 февраля 2018 в 04:02 Источник Поделиться

Единственная вещь, которую я заметил, заключается в том, что retry потенциально непоследовательное поведение.

Позвольте мне объяснить:

def retry(func, *func_args, **kwargs):
retry_count = kwargs.get("retry_count", 5)
delay = kwargs.get("delay", 5)
while retry_count > 1:
if func(*func_args):
return
log.debug("waiting for %s seconds before retyring again")
sleep(delay)
retry_count = retry_count - 1

return func(*func_args)


Если func при проверки внутри самой while,
None будут возвращены. С другой стороны, если она будет успешной
снаружи на whileон будет возвращать все func возвращает (в вашем примере True). Вы не хотите иметь что..

Поэтому я хотел бы предложить небольшое изменение кода:

def retry(func, *func_args, **kwargs):
retry_count = kwargs.get("retry_count", 5)
delay = kwargs.get("delay", 5)
for _ in range(retry_count): # all tries happen in the loop
if func(*func_args):
return True # if we succeed we return True
log.debug("waiting for %s seconds before retyring again")
sleep(delay)
return False # if we did not, we return False


Вы можете сделать попышнее, если вы хотите купить subsituting выше for петли с этим:

for _ in range(retry_count):
res = func(*func_args) or log.debug("waiting for %s seconds before retyring again")
if res is None:
sleep(delay)
else:
return True

Обратите внимание, что я assuiming здесь log.debug возвращает None но это не так важно, покуда он не вернется True.

12
ответ дан 28 февраля 2018 в 04:02 Источник Поделиться

Теория

Ваш retry функции очень похожа на структуру any.

Сохраняя только необходимые, вы могли бы написать retry как :

any(func(*func_args) for _ in range(count))

Код

Если вы хотите kwargs, log и sleepможно написать:

def retry(func, *func_args, **kwargs):
count = kwargs.pop("count", 5)
delay = kwargs.pop("delay", 5)
return any(func(*func_args, **kwargs)
or log.debug("waiting for %s seconds before retyring again" % delay)
or time.sleep(delay)
for _ in range(count))

Примечания


  • log.debug и time.sleep как ложь, так func or log or time - истина, если и только если func - истина.

  • dict.pop нужен для извлечения count и delay от kwargs. Они будут вам переданы func в противном случае.

Полный код

import time
import pwd
import logging as log

def is_user(username):
try:
pwd.getpwnam(username)
log.info("User %s exist" % username)
return True
except KeyError:
log.error("User %s does not exist." % username)
return False

def retry(func, *func_args, **kwargs):
count = kwargs.pop("count", 5)
delay = kwargs.pop("delay", 5)
return any(func(*func_args, **kwargs)
or log.debug("waiting for %s seconds before retyring again" % delay)
or time.sleep(delay)
for _ in range(count))

retry(is_user, 'username', count=10, delay=0.5)

2
ответ дан 28 февраля 2018 в 09:02 Источник Поделиться

Вот некоторые проблемы с вашей текущей настройки:


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

  2. Использование повторов, вы должны позвонить retry(is_user_exist, username, ...). Это делает это тяжелее, чтобы избежать повторения.

  3. Вы можете в конечном итоге с той же отладочные появляться в журналах 5 раз подряд.

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

Я предлагаю декоратор, как указано ниже. Я писал этот декоратор некоторое время назад и с удовольствием используют его в нескольких местах. Идея похожа на ответ Dannnno, но я только повторить после исключения и не обращать внимание на возвращаемые значения.

def retry(num_attempts=3, exception_class=Exception, log=None, sleeptime=1):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for i in range(num_attempts):
try:
return func(*args, **kwargs)
except exception_class as e:
if i == num_attempts - 1:
raise
else:
if log:
log.error('Failed with error %r, trying again', e)
sleep(sleeptime)

return wrapper

return decorator

Вот некоторые использования в реальном коде:

from requests.exceptions import ConnectionError

@retry(5, ConnectionError, log)
def request(self, method, url='', **kwargs):
...

Вот еще один пример, где я повторить только в месте вызова, а не изменить определение функции:

retry(5, Exception, log)(ftp.get)(path, destination)

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

if is_user_exist(username):
process_existing_user()
else:
process_nonexistent_user()

становится:

try:
retry(5, KeyError, log)(pwd.getpwnam)(username)
except KeyError:
process_nonexistent_user()
else:
process_existing_user()

Если у вас есть другие функции, которые вы хотите, чтобы повторить, когда условие ложно, что не предполагают исключения, вы могли бы явным образом вызывать исключение:

class StuffNotFound:
pass

@retry(exception_class=StuffNotFound)
def process_stuff():
stuff = get_stuff():
if not stuff:
raise StuffNotFound()
process(stuff)

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

1
ответ дан 1 марта 2018 в 01:03 Источник Поделиться

Вы можете заменить цикл while и retry_count обратный отсчет с помощью простого цикла for через диапазон()

def retry(func, *func_args, **kwargs):
retry_count = kwargs.get("retry_count", 5)
delay = kwargs.get("delay", 5)
for _ in range(retry_count):
if func(*func_args):
return
log.debug("waiting for %s seconds before retyring again")
sleep(delay)

return func(*func_args)

-2
ответ дан 28 февраля 2018 в 05:02 Источник Поделиться