如何在 Python 的 for 循环中创建超时？答案

【问题标题】：How do you create a timeout in a for-loop in Python?如何在 Python 的 for 循环中创建超时？
【发布时间】：2021-01-21 16:44:29
【问题描述】：

我有一个从 API 检索数据的 for 循环：

app = WebService()
for i in items:
    result = app.request(item)

我想创建一个超时，这样，如果 app.request 阻塞调用花费的时间太长，循环将跳过它并转到下一项。

我已经阅读了一些使用while 循环创建计时器的方法，但我相信在我的情况下，我无法在for 循环内创建一个while 循环，其中continue 子句适用于for 循环...那么，我该怎么做呢？

很遗憾，API 没有提供创建超时的方法。它不是对 REST API 的 HTTP 调用。

【问题讨论】：

是否以及如何中止呼叫取决于执行的实际操作。如果不知道 app.request 是/做什么，就不可能正确回答这个问题。
Make a blocking function timeout in Python 会回答您的问题吗？还是Timeout on a function call?
您能列出您正在使用的 API 吗？假设app.request 是一个同步套接字函数，除了直接干扰已经打开连接的套接字之外，没有“非脏”的方法可以做到这一点。你可以创建另一个线程并设置一个超时，然后你可以做一些会导致异常的事情，导致app.request停止。

标签： python

【解决方案1】：

基于this answer。

这是decorator 的一个很好的用例。当您想要包装具有附加功能的函数或类方法时，装饰器模式很有用。

这相当于如何使用 Python 的 signal 库执行 Python 3 超时装饰器。

获得装饰器后，将 app.request 包装在 decorated 函数中，并使用 try-except 处理异常。

# main.py
import signal

DEBUG = True

# Custom exception. In general, it's a good practice.
class TimeoutError(Exception):
    def __init__(self, value = "Timed Out"):
        self.value = value
    def __str__(self):
        return repr(self.value)

# This is the decorator itself.
# Read about it here: https://pythonbasics.org/decorators/
# Basically, it's a function that receives another function and a time parameter, i.e. the number of seconds.
# Then, it wraps it so that it raises an error if the function does not
# return a result before `seconds_before_timeout` seconds
def timeout(seconds_before_timeout):
  def decorate(f):
    if DEBUG: print("[DEBUG]: Defining decorator handler and the new function")
    if DEBUG: print(f"[DEBUG]: Received the following function >> `{f.__name__}`")
    def handler(signum, frame):
      raise TimeoutError()
    def new_f(*args, **kwargs):
      # Verbatim from Python Docs
      # > The signal.signal() function allows defining custom handlers to be executed
      #   when a signal is received.
      if DEBUG: print(f"[DEBUG]: in case of ALARM for function `{f.__name__}`, I'll handle it with the... `handler`")
      old = signal.signal(signal.SIGALRM, handler)

      # See https://docs.python.org/3/library/signal.html#signal.alarm
      if DEBUG: print(f"[DEBUG]: setting an alarm for {seconds_before_timeout} seconds.")
      signal.alarm(seconds_before_timeout)
      try:
          if DEBUG: print(f"[DEBUG]: executing `{f.__name__}`...")
          result = f(*args, **kwargs)
      finally:
          # reinstall the old signal handler
          signal.signal(signal.SIGALRM, old)
          # Cancel alarm. 
          # See: https://docs.python.org/3/library/signal.html#signal.alarm
          signal.alarm(0)
      return result
    
    new_f.__name__ = f.__name__
    return new_f
  return decorate

import time

@timeout(5)
def mytest():
    for i in range(1,10):
      interval = 2
      if DEBUG: print("[DEBUG]: waiting 2 seconds... on purpose")
      time.sleep(interval)
      print("%d seconds have passed" % (interval * i))

if __name__ == '__main__':
  if DEBUG: print("[DEBUG]: Starting program")
  mytest()

您可以在this repl.it上快速尝试

此外，如果您不想在函数内“隐藏”API 调用，您可以进行依赖倒置。也就是说，您的装饰函数并不特别依赖于任何 request 函数实现。为此，您将函数本身作为参数接收。见下文：

# simple decoration
@timeout(5)
def make_request(item):
    # assuming `app` is defined
    return app.request(item)

# dependency inversion

@timeout(5)
def make_request(request_handler, item):
    return request_handler(item)

# and then in your loop...
for i in items:
    make_request(app.request, item)

【讨论】：