Puedes usar multiprocessing.Process para hacer exactamente eso.

Código

import multiprocessing
import time

# bar
def bar():
    for i in range(100):
        print "Tick"
        time.sleep(1)

if __name__ == '__main__':
    # Start bar as a process
    p = multiprocessing.Process(target=bar)
    p.start()

    # Wait for 10 seconds or until process finishes
    p.join(10)

    # If thread is still active
    if p.is_alive():
        print "running... let's kill it..."

        # Terminate
        p.terminate()
        p.join()

¿Cómo llamo a la función o en qué la envuelvo para que si tarda más de 5 segundos el script la cancele?

Publiqué un gist que resuelve esta pregunta/problema con un decorador y un threading.Timer. Aquí está con una crisis.

Importaciones y configuraciones para compatibilidad

Se probó con Python 2 y 3. También debería funcionar bajo Unix / Linux y Windows.

Primero las importaciones. Estos intentan mantener el código consistente independientemente de la versión de Python:

from __future__ import print_function
import sys
import threading
from time import sleep
try:
    import thread
except ImportError:
    import _thread as thread

Use código independiente de la versión:

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Ahora hemos importado nuestra funcionalidad de la biblioteca estándar.

exit_after decorador

A continuación necesitamos una función para terminar el main() desde el hilo hijo:

def quit_function(fn_name):
    # print to stderr, unbuffered in Python 2.
    print('{0} took too long'.format(fn_name), file=sys.stderr)
    sys.stderr.flush() # Python 3 stderr is likely buffered.
    thread.interrupt_main() # raises KeyboardInterrupt

Y aquí está el decorador mismo:

def exit_after(s):
    '''
    use as decorator to exit process if 
    function takes longer than s seconds
    '''
    def outer(fn):
        def inner(*args, **kwargs):
            timer = threading.Timer(s, quit_function, args=[fn.__name__])
            timer.start()
            try:
                result = fn(*args, **kwargs)
            finally:
                timer.cancel()
            return result
        return inner
    return outer

Uso

Y aquí está el uso que responde directamente a su pregunta sobre salir después de 5 segundos!:

@exit_after(5)
def countdown(n):
    print('countdown started', flush=True)
    for i in range(n, -1, -1):
        print(i, end=', ', flush=True)
        sleep(1)
    print('countdown finished')

Demo:

>>> countdown(3)
countdown started
3, 2, 1, 0, countdown finished
>>> countdown(10)
countdown started
10, 9, 8, 7, 6, countdown took too long
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 6, in countdown
KeyboardInterrupt

La segunda llamada a la función no finalizará, en su lugar el proceso debe salir con un seguimiento!

KeyboardInterrupt no siempre detiene un hilo para dormir

Tenga en cuenta que la suspensión no siempre será interrumpida por una interrupción del teclado, en Python 2 en Windows, por ejemplo:

@exit_after(1)
def sleep10():
    sleep(10)
    print('slept 10 seconds')

>>> sleep10()
sleep10 took too long         # Note that it hangs here about 9 more seconds
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 3, in sleep10
KeyboardInterrupt

Tampoco es probable que interrumpa el código que se ejecuta en extensiones a menos que verifique explícitamente PyErr_CheckSignals(), consulte Cython, Python y KeyboardInterrupt ignorado

Evitaría dormir un hilo más de un segundo, en cualquier caso, eso es un eon en tiempo de procesador.

¿Cómo llamo a la función o en qué la envuelvo para que si tarda más de 5 segundos el script la cancele y haga otra cosa?

Para atraparlo y hacer otra cosa, puede capturar la interrupción del teclado.

>>> try:
...     countdown(10)
... except KeyboardInterrupt:
...     print('do something else')
... 
countdown started
10, 9, 8, 7, 6, countdown took too long
do something else

Tengo una propuesta diferente que es una función pura (con la misma API que la sugerencia de threading) y parece funcionar bien (basado en sugerencias sobre este hilo)

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    import signal

    class TimeoutError(Exception):
        pass

    def handler(signum, frame):
        raise TimeoutError()

    # set the timeout handler
    signal.signal(signal.SIGALRM, handler) 
    signal.alarm(timeout_duration)
    try:
        result = func(*args, **kwargs)
    except TimeoutError as exc:
        result = default
    finally:
        signal.alarm(0)

    return result

Me encontré con este hilo al buscar una llamada de tiempo de espera en las pruebas unitarias. No encontré nada simple en las respuestas o paquetes de 3rd party, así que escribí el decorador a continuación, puede caer directamente en el código:

import multiprocessing.pool
import functools

def timeout(max_timeout):
    """Timeout decorator, parameter in seconds."""
    def timeout_decorator(item):
        """Wrap the original function."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Closure for function."""
            pool = multiprocessing.pool.ThreadPool(processes=1)
            async_result = pool.apply_async(item, args, kwargs)
            # raises a TimeoutError if execution exceeds max_timeout
            return async_result.get(max_timeout)
        return func_wrapper
    return timeout_decorator

Entonces es tan simple como esto para timeout una prueba o cualquier función que te gusta:

@timeout(5.0)  # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
    ...

Hay muchas sugerencias, pero ninguna usando concurrente.futuros, que creo que es la forma más legible de manejar esto.

from concurrent.futures import ProcessPoolExecutor

# Warning: this does not terminate function if timeout
def timeout_five(fnc, *args, **kwargs):
    with ProcessPoolExecutor() as p:
        f = p.submit(fnc, *args, **kwargs)
        return f.result(timeout=5)

Súper fácil de leer y mantener.

Hacemos un pool, enviamos un solo proceso y luego esperamos hasta 5 segundos antes de generar un TimeoutError que puedas atrapar y manejar como lo necesites.

Nativo de python 3.2+ y backported a 2.7 (pip install futures).

Cambiar entre hilos y procesos es tan simple como reemplazar ProcessPoolExecutor con ThreadPoolExecutor.

Si desea terminar el Proceso en el tiempo de espera, le sugeriría buscar en Pebble.

El paquete stopit, encontrado en pypi, parece manejar bien los tiempos de espera.

Me gusta el decorador @stopit.threading_timeoutable, que agrega un parámetro timeout a la función decorada, que hace lo que espera, detiene la función.

Comprobarlo en pypi: https://pypi.python.org/pypi/stopit

#!/usr/bin/python2
import sys, subprocess, threading
proc = subprocess.Popen(sys.argv[2:])
timer = threading.Timer(float(sys.argv[1]), proc.terminate)
timer.start()
proc.wait()
timer.cancel()
exit(proc.returncode)

Necesitaba interrupciones cronometradas (que SIGALARM no puede hacer) que no se bloquearán por el tiempo.sueño (que el enfoque basado en hilos no puede hacer). Terminé copiando y modificando ligeramente el código desde aquí: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/

El código en sí:

#!/usr/bin/python

# lightly modified version of http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/


"""alarm.py: Permits multiple SIGALRM events to be queued.

Uses a `heapq` to store the objects to be called when an alarm signal is
raised, so that the next alarm is always at the top of the heap.
"""

import heapq
import signal
from time import time

__version__ = '$Revision: 2539 $'.split()[1]

alarmlist = []

__new_alarm = lambda t, f, a, k: (t + time(), f, a, k)
__next_alarm = lambda: int(round(alarmlist[0][0] - time())) if alarmlist else None
__set_alarm = lambda: signal.alarm(max(__next_alarm(), 1))


class TimeoutError(Exception):
    def __init__(self, message, id_=None):
        self.message = message
        self.id_ = id_


class Timeout:
    ''' id_ allows for nested timeouts. '''
    def __init__(self, id_=None, seconds=1, error_message='Timeout'):
        self.seconds = seconds
        self.error_message = error_message
        self.id_ = id_
    def handle_timeout(self):
        raise TimeoutError(self.error_message, self.id_)
    def __enter__(self):
        self.this_alarm = alarm(self.seconds, self.handle_timeout)
    def __exit__(self, type, value, traceback):
        try:
            cancel(self.this_alarm) 
        except ValueError:
            pass


def __clear_alarm():
    """Clear an existing alarm.

    If the alarm signal was set to a callable other than our own, queue the
    previous alarm settings.
    """
    oldsec = signal.alarm(0)
    oldfunc = signal.signal(signal.SIGALRM, __alarm_handler)
    if oldsec > 0 and oldfunc != __alarm_handler:
        heapq.heappush(alarmlist, (__new_alarm(oldsec, oldfunc, [], {})))


def __alarm_handler(*zargs):
    """Handle an alarm by calling any due heap entries and resetting the alarm.

    Note that multiple heap entries might get called, especially if calling an
    entry takes a lot of time.
    """
    try:
        nextt = __next_alarm()
        while nextt is not None and nextt <= 0:
            (tm, func, args, keys) = heapq.heappop(alarmlist)
            func(*args, **keys)
            nextt = __next_alarm()
    finally:
        if alarmlist: __set_alarm()


def alarm(sec, func, *args, **keys):
    """Set an alarm.

    When the alarm is raised in `sec` seconds, the handler will call `func`,
    passing `args` and `keys`. Return the heap entry (which is just a big
    tuple), so that it can be cancelled by calling `cancel()`.
    """
    __clear_alarm()
    try:
        newalarm = __new_alarm(sec, func, args, keys)
        heapq.heappush(alarmlist, newalarm)
        return newalarm
    finally:
        __set_alarm()


def cancel(alarm):
    """Cancel an alarm by passing the heap entry returned by `alarm()`.

    It is an error to try to cancel an alarm which has already occurred.
    """
    __clear_alarm()
    try:
        alarmlist.remove(alarm)
        heapq.heapify(alarmlist)
    finally:
        if alarmlist: __set_alarm()

Y un ejemplo de uso:

import alarm
from time import sleep

try:
    with alarm.Timeout(id_='a', seconds=5):
        try:
            with alarm.Timeout(id_='b', seconds=2):
                sleep(3)
        except alarm.TimeoutError as e:
            print 'raised', e.id_
        sleep(30)
except alarm.TimeoutError as e:
    print 'raised', e.id_
else:
    print 'nope.'

Aquí hay una ligera mejora a la solución dada basada en subprocesos.

El siguiente código admite excepciones :

def runFunctionCatchExceptions(func, *args, **kwargs):
    try:
        result = func(*args, **kwargs)
    except Exception, message:
        return ["exception", message]

    return ["RESULT", result]


def runFunctionWithTimeout(func, args=(), kwargs={}, timeout_duration=10, default=None):
    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = default
        def run(self):
            self.result = runFunctionCatchExceptions(func, *args, **kwargs)
    it = InterruptableThread()
    it.start()
    it.join(timeout_duration)
    if it.isAlive():
        return default

    if it.result[0] == "exception":
        raise it.result[1]

    return it.result[1]

Invocándolo con un tiempo de espera de 5 segundos:

result = timeout(remote_calculate, (myarg,), timeout_duration=5)

Podemos usar señales para lo mismo. Creo que el siguiente ejemplo será útil para usted. Es muy simple en comparación con los hilos.

import signal

def timeout(signum, frame):
    raise myException

#this is an infinite loop, never ending under normal circumstances
def main():
    print 'Starting Main ',
    while 1:
        print 'in main ',

#SIGALRM is only usable on a unix platform
signal.signal(signal.SIGALRM, timeout)

#change 5 to however many seconds you need
signal.alarm(5)

try:
    main()
except myException:
    print "whoops"

Excelente, fácil de usar y confiable PyPI project tiempo de espera-decorador (https://pypi.org/project/timeout-decorator/)

Instalación:

pip install timeout-decorator

Uso :

import time
import timeout_decorator

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

if __name__ == '__main__':
    mytest()

timeout-decorator no funciona en el sistema Windows, ya que Windows no es compatible con signal bien.

Si utiliza timeout-decorator en el sistema windows obtendrá lo siguiente

AttributeError: module 'signal' has no attribute 'SIGALRM'

Algunos sugirieron usar use_signals=False pero no funcionó para mí.

El autor @bitranox creó el siguiente paquete:

pip install https://github.com/bitranox/wrapt-timeout-decorator/archive/master.zip

Ejemplo de código:

import time
from wrapt_timeout_decorator import *

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

def main():
    mytest('starting')


if __name__ == '__main__':
    main()

Da la siguiente excepción:

TimeoutError: Function mytest timed out after 5 seconds