Таймаут при вызове функции

Вы можете использовать multiprocessing.Process для выполнения этой задачи.

Код

import multiprocessing
import time

# Функция bar
def bar():
    for i in range(100):
        print("Tick")
        time.sleep(1)

if __name__ == '__main__':
    # Запуск функции bar в отдельном процессе
    p = multiprocessing.Process(target=bar)
    p.start()

    # Ожидание 10 секунд или пока процесс не завершится
    p.join(10)

    # Проверка, все еще ли процесс активен
    if p.is_alive():
        print("Работает... давайте остановим его...")

        # Завершение процесса - может не сработать, если процесс застрял
        p.terminate()
        # ИЛИ Убить - сработает в любом случае, но процесс не завершится корректно
        # p.kill()

        p.join()

Этот пример демонстрирует, как создать и управлять отдельным процессом в Python. Используя multiprocessing, вы можете запускать функции параллельно и при необходимости их завершать. Обратите внимание, что использование terminate() может не сработать, если процесс застрял, в то время как kill() обязательно завершит процесс, но не даст ему возможности корректно завершиться.

У меня есть другое предложение, которое является чистой функцией (при этом сохраняет тот же API, что и предложение с многопоточностью) и, судя по обсуждениям в этой ветке, работает нормально:

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

    class TimeoutError(Exception):
        pass

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

    # Устанавливаем обработчик таймаута
    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

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

Я наткнулся на эту тему, когда искал возможность установки таймаута для юнит-тестов. Не нашел ничего простого в ответах или сторонних пакетах, поэтому написал декоратор, который можно просто вставить в код:

import multiprocessing.pool
import functools

def timeout(max_timeout):
    """Декоратор таймаута, параметр в секундах."""
    def timeout_decorator(item):
        """Оборачивает оригинальную функцию."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Замыкание для функции."""
            pool = multiprocessing.pool.ThreadPool(processes=1)
            async_result = pool.apply_async(item, args, kwargs)
            # вызовет TimeoutError, если выполнение превысит max_timeout
            return async_result.get(max_timeout)
        return func_wrapper
    return timeout_decorator

После этого всё просто: достаточно установить таймаут для теста или любой другой функции, как показано ниже:

@timeout(5.0)  # если выполнение займет больше 5 секунд, выбросит TimeoutError
def test_base_regression(self):
    ...

Этот подход легко интегрируется в ваши тесты и позволяет контролировать время выполнения.

Пакет stopit, доступный на pypi, отлично справляется с тайм-аутами.

Мне нравится декоратор @stopit.threading_timeoutable, который добавляет параметр timeout к декорируемой функции. Он работает так, как ожидается: останавливает выполнение функции при превышении указанного времени.

Посмотрите его на pypi: https://pypi.python.org/pypi/stopit

Похоже, вы ищете информацию о проекте timeout-decorator на PyPi. Это отличный и простой в использовании инструмент, который позволяет устанавливать тайм-ауты для функций в Python.

Вот как вы можете его установить:

pip install timeout-decorator

После установки вы можете использовать его следующим образом:

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

В этом примере функция mytest будет автоматически прервана, если выполнение займет больше 5 секунд. Это делает timeout-decorator полезным для предотвращения зависания функций, которые могут выполняться дольше, чем ожидается. Не забудьте обрабатывать возможные исключения, так как при превышении времени ожидания будет вызвано исключение timeout_decorator.TimeoutError.