Ошибка превышения времени ожидания в коде на Python

Расширенные примеры работы с тайм-аутами в Python

Тайм-аут с использованием модуля functools и частичного применения

Можно создать декоратор, реализующий тайм-аут для любой функции с помощью signal (только Unix). Пример с поддержкой целочисленного и дробного времени:

import signal
from functools import wraps

class TimeoutError(Exception):
    pass

def timeout(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            def handler(signum, frame):
                raise TimeoutError(f"Функция {func.__name__} превысила лимит {seconds} сек")
            old_handler = signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                signal.alarm(0)
                signal.signal(signal.SIGALRM, old_handler)
        return wrapper
    return decorator

@timeout(3)
def slow_computation():
    import time
    time.sleep(5)
    return 42

try:
    print(slow_computation())
except TimeoutError as e:
    print(e)

Функция slow_computation превысила лимит 3 сек

Асинхронный тайм-аут с отменой корутины и побочными эффектами

Пример демонстрирует, что задача после тайм-аута отменяется, и можно выполнить очистку:

import asyncio

async def resource_intensive_task():
    try:
        print("Задача запущена")
        await asyncio.sleep(10)
        print("Задача завершена (не должно быть)")
    except asyncio.CancelledError:
        print("Задача отменена, освобождаем ресурсы")
        # Здесь можно закрыть соединения, файлы и т.д.
        raise

async def main():
    task = asyncio.create_task(resource_intensive_task())
    try:
        await asyncio.wait_for(task, timeout=5)
    except asyncio.TimeoutError:
        print("Основная программа: тайм-аут")
    # Ждем, чтобы убедиться, что задача действительно отменена
    await asyncio.sleep(1)
    print("Конец main")

asyncio.run(main())

Задача запущена
Задача отменена, освобождаем ресурсы
Основная программа: тайм-аут
Конец main

Тайм-аут для HTTP с использованием aiohttp

Библиотека aiohttp позволяет устанавливать тайм-аут как на соединение, так и на чтение:

import aiohttp
import asyncio

async def fetch_with_timeout():
    timeout = aiohttp.ClientTimeout(total=5, connect=2, sock_read=3)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get('https://httpbin.org/delay/10') as response:
                return await response.text()
        except asyncio.TimeoutError:
            return "Тайм-аут aiohttp"
        except aiohttp.ClientError as e:
            return f"Ошибка клиента: {e}"

result = asyncio.run(fetch_with_timeout())
print(result)

Тайм-аут aiohttp

Тайм-аут при работе с контекстными менеджерами (например, открытие файла)

Хотя стандартный open не поддерживает тайм-аут, можно обернуть его в поток или процесс. Пример с использованием multiprocessing и Queue:

import multiprocessing
def read_file(filename, queue):
    try:
        with open(filename, 'r') as f:
            data = f.read()
        queue.put(data)
    except Exception as e:
        queue.put(e)

def timed_read(filename, timeout):
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(target=read_file, args=(filename, queue))
    process.start()
    process.join(timeout)
    if process.is_alive():
        process.terminate()
        process.join()
        raise TimeoutError(f"Чтение файла {filename} превысило {timeout} сек")
    result = queue.get()
    if isinstance(result, Exception):
        raise result
    return result

try:
    content = timed_read('/dev/random', 2)  # /dev/random может блокироваться
    print(content[:100])
except TimeoutError as e:
    print(e)

Чтение файла /dev/random превысило 2 сек

Комбинирование тайм-аутов: обертка для внешних библиотек

Иногда требуется установить тайм-аут на вызов C-расширения, которое не поддерживает отмену. В таких случаях помогает запуск в отдельном процессе:

import multiprocessing
import time

def call_c_extension(result_dict):
    # Имитация вызова C-функции, которая может зависнуть
    time.sleep(10)
    result_dict['result'] = 42

manager = multiprocessing.Manager()
result = manager.dict()
p = multiprocessing.Process(target=call_c_extension, args=(result,))
p.start()
p.join(5)
if p.is_alive():
    p.terminate()
    p.join()
    raise TimeoutError("C-расширение не вернулось вовремя")
print(result.get('result', 'Нет результата'))

... (процесс убит) TimeoutError: C-расширение не вернулось вовремя

Тайм-аут в Jupyter Notebook (ячейки)

Для интерактивных сред можно использовать %%timeout магию (требуется пакет ipython-autotime или собственный обработчик). Пример собственного декоратора для ячеек:

from IPython.core.magic import register_cell_magic
import signal

@register_cell_magic
def cell_timeout(line, cell):
    try:
        seconds = int(line)
    except ValueError:
        seconds = 5
    
    def handler(signum, frame):
        raise TimeoutError("Ячейка превысила время выполнения")
    
    old_handler = signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)
    try:
        exec(cell)
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

# Использование в ячейке:
# %%cell_timeout 2
# import time
# time.sleep(10)

(в случае превышения: TimeoutError: Ячейка превысила время выполнения)

Тайм-аут при использовании библиотеки socket (подробный пример с таймером на чтение)

Пример суммарного тайм-аута на чтение данных заданного размера:

import socket
import time

def recv_timeout(sock, n, timeout_seconds):
    sock.settimeout(timeout_seconds)  # тайм-аут на каждый recv
    data = b''
    start = time.time()
    while len(data) < n:
        remaining = n - len(data)
        if time.time() - start > timeout_seconds:
            raise socket.timeout("Суммарное время чтения превышено")
        try:
            chunk = sock.recv(remaining)
        except socket.timeout:
            continue  # просто повторяем попытку
        if not chunk:
            break
        data += chunk
    return data

sock = socket.socket()
sock.connect(('httpbin.org', 80))
sock.sendall(b'GET /bytes/1000 HTTP/1.1\r\nHost: httpbin.org\r\n\r\n')
# Пропускаем заголовки, читаем тело
sock.recv(4096)  # заголовки (можно отдельно)
try:
    body = recv_timeout(sock, 1000, 5)
    print(f"Прочитано {len(body)} байт")
except socket.timeout as e:
    print(e)
finally:
    sock.close()

Прочитано 1000 байт (или сообщение о тайм-ауте)