How do you create a timeout in a for-loop in Python?

Question

I have a for loop that retrieves data from an API:

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

I want to create a timeout so that, if the app.request blocking call takes too long, the loop will skip it and go to the next item.

I have read some ways of creating a timer by using a while loop, but I believe in my case I cannot create a while loop inside the for loop with a continue clause that would apply to the for loop... So, how can I do this?

Unfortunately, the API doesn't provide a way to create a timeout. It is not an HTTP call to a REST API.

question from:https://stackoverflow.com/questions/65831922/how-do-you-create-a-timeout-in-a-for-loop-in-python

Answer 1

Based on this answer.

This is a good use-case for a decorator. The decorator pattern is useful when you want to wrap a function or class method with additional functionality.

This is the equivalent of how to do a Python 3 timeout decorator using the Python's signal library.

Once you have the decorator, wrap your app.request in a decorated function and handle the exception with a 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()

You can try it fast on this repl.it

---

Additionally, if you do not want to "hide" the API call inside the function, you can do dependency inversion. That is, your decorated function does not depend on any request function implementation in particular. To do this, you receive the function itself as an argument. See below:

# 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)