How do I timeout and replay a command line in Python?

Question

I have a request for information that simply does not receive information and stands still, so forward executed and I have the answer. Then, how to have the command line be canceled after a time limit and re-exported until an event occurs?

Example:

def atrasado():
    import random
    import time
    a = random.random()
    time.sleep(a*10)
    if a < 0.5:
        print('Foi')
        print(a*10)
    else:
        print('erro')
        print(a*10)

So how to give timeout in atrasado() if it takes longer than 5 seconds and restarts the function?

There is always the situation of atrasado() take longer than 5 seconds despite having been reboot.

Answer 1

As commented, you can use the reply of Timeout in Python input function along with a repeat loop, so re-run the function when timeout occurs.

@timeout(seconds=5)
def read_user_name():
    name = input('Qual é o seu nome? ')
    print('Seja bem-vindo,', name)

attempts = 5

while attempts:
    try:
        read_user_name()
        break
    except Exception as error:
        print(error)
        attempts -= 1
else:
    print('Erro: acabaram suas tentativas')

^{See working on repl it.}

Answer 2

One solution to your problem is to use the module asyncio.
asyncio is a library for writing concurrent asynchronous code using syntax async/await.

asyncio operates on corroding which are a more generalized form of subroutines. Subroutines have input started at one point, and output at another point. Corroding can enter, exit, and continue at many different points and corroding are declared with the syntax async/await.

Perform a corroding with:

• asyncio.run() executes and returns the result of corroutine.
• asyncio.create_task() agenda and implementation of a corroding involves her in a Task returning him.
• wait awaiting the return of a corroding. Can only be used inside another corrosion.

As an example two will be created corroding:

• demorada() which is a corrosion that may or may not carry 10s to be executed, is a random choice decided on the line await asyncio.sleep(random.choice((0,10))) holding or not the processing by 10s. Returns a random number in the range [0,65536[.
• main(timeout=1) is top-level corrosion. Creates an activity for corrosion demorada() and is on hold with timeout through the method asyncio.wait_for(aw, timeout) making an exception asyncio.TimeoutError if the running time of the activity lasts longer than timeout.

Example:

import asyncio
import random

async def demorada():
    await asyncio.sleep(random.choice((0,10)))            #Aleatoriamente sorteia se demora ou não 10s
    return random.randint(0, 65536)                       #Retorna no internalo [0,65536[
    
async def main(timeout=1):
   task = asyncio.create_task(demorada())                 #Agenda a corrotina demorada() e retorna um task. 
   try:
       await asyncio.wait_for(task, timeout)              #Aguarda a execução da Task com timeout.
   except asyncio.TimeoutError:
       task.cancel()                                      #Cancela a atividade.
       print("A função demorou demais para responder.")   #Imprime a mensagem.
   else:
       print(f"A função retornou {task.result()}")        #Imprime a mensagem com resultado de demorada()

#Faz 10 chamadas de main() com timeout 5s
for _ in range(10):
    print(">>> ",end="", flush="True")          
    asyncio.run(main(5))

Resulting:

>>> A função retornou 43258
>>> A função demorou demais para responder.
>>> A função retornou 14525
>>> A função retornou 6355
>>> A função demorou demais para responder.
>>> A função retornou 39265
>>> A função demorou demais para responder.
>>> A função retornou 39502
>>> A função demorou demais para responder.
>>> A função retornou 50657

Applying this information to the example of the question just create a loop around the activity and reiterate case of timeout or check its result in a timely manner.

import asyncio
import random

async def demorada():
    await asyncio.sleep(random.randint(0,10))            
    return random.randint(0, 65536)                       

async def main(timeout=1):
   #Abre um loop de duração indeterminada...
   while True:
       task = asyncio.create_task(demorada())          #Agenda a execução da atividade demorada
       print('Fazendo uma tentativa...')               #Imprime mensagem
       #Abre um bloco de tratamento de exceções...
       try:
           await asyncio.wait_for(task, timeout)       #Aguarda a execução da Task com timeout.       
       except asyncio.TimeoutError:                    #Havendo exceção de timeout...
           task.cancel()                               #Cancela a atividade.
           print('Time out...')                        #Imprime mensagem permanecendo no loop.
       else:
           return task.result()                        #Retorna o resultado da atividade.


req = asyncio.run(main(5))
print(f'Resultado da chamada {req}')

Resulting:

Fazendo uma tentativa...
Time out...
Fazendo uma tentativa...
Time out...
Fazendo uma tentativa...
Resultado da chamada 20864

Answer 3

You can use the Retry Pattern to perform its function again.

Retry is a Pattern used in various languages to redo a function call if an error occurs, in your implementation you can specify the number of attempts your function will have to perform after this number of attempts if still persist the error it will be terminated and an exception will be thrown

Whenever we assume that an unexpected response - or no response in this case - can be corrected by submitting the request again, using the repeat pattern may help.

from retry import retry
@retry()
def make_trouble():
    '''Retry until succeed'''

There are also other libraries like timeout-Decorator which you can use to limit the time of your function.

timeout-Decorator is also a Pattern used to limit the running time of a function

Timeout Pattern is quite simple and many HTTP clients have a default timeout set. The goal is to avoid unlimited waiting times for responses and thus treat each request as a failure when no response has been received within the time limit.

import time
import timeout_decorator

@timeout_decorator.timeout(5)
def mytest():
    print("Start")
    for i in range(1,10):
        time.sleep(1)
        print("{} seconds have passed".format(i))

if __name__ == '__main__':
    mytest()

Apart from this two there are still some Pattern that comes the trough from time to time as the Circuit Breaker, fallback

Its function would be as follows:

from retry import retry
import random
import time


@retry(tries=3, delay=2)
def atrasado():
    a = random.random()
    time.sleep(a*10)
    if a < 0.5:
        print('Foi')
        print(a*10)
    else:
        print("Nova Tetativa")        
        raise Exception("Erro de tempo")

if __name__ == "__main__":
    atrasado()