2.7. @aiomisc.timeout

Decorator that ensures the execution time limit for the decorated function is met.

from aiomisc import timeout

@timeout(1)
async def bad_func():
    await asyncio.sleep(2)

2.7.1. What happens on timeout

When the decorated function exceeds the specified time limit, an asyncio.TimeoutError exception is raised. The underlying coroutine is cancelled automatically.

import asyncio
from aiomisc import timeout

@timeout(1)
async def slow_operation():
    await asyncio.sleep(10)
    return "completed"

async def main():
    try:
        result = await slow_operation()
    except asyncio.TimeoutError:
        print("Operation timed out!")

asyncio.run(main())

2.7.2. Handling TimeoutError

You should always handle asyncio.TimeoutError when calling functions decorated with @timeout, especially in production code:

import asyncio
from aiomisc import timeout

@timeout(5)
async def fetch_data():
    # Simulate network request
    await asyncio.sleep(10)
    return {"data": "value"}

async def main():
    try:
        data = await fetch_data()
        print(f"Got data: {data}")
    except asyncio.TimeoutError:
        print("Request timed out, using default value")
        data = {"data": "default"}

asyncio.run(main())

2.7.3. Comparison with asyncio.wait_for

The @timeout decorator provides similar functionality to asyncio.wait_for, but with a cleaner decorator-based syntax:

asyncio.wait_for

@aiomisc.timeout

async def fetch():
    await asyncio.sleep(10)

# At call site:
await asyncio.wait_for(
    fetch(), timeout=5
)
@timeout(5)
async def fetch():
    await asyncio.sleep(10)

# At call site:
await fetch()

The decorator approach is useful when:

  • You want to enforce a timeout for all calls to a function

  • The timeout is a property of the function itself, not the caller

  • You want cleaner call-site code

2.7.4. Examples with different values

The timeout value is specified in seconds and can be an integer or float:

from aiomisc import timeout

# 100 milliseconds timeout
@timeout(0.1)
async def quick_check():
    await asyncio.sleep(0.05)
    return True

# 30 seconds timeout for long operations
@timeout(30)
async def long_operation():
    await asyncio.sleep(25)
    return "done"

# 2.5 seconds timeout
@timeout(2.5)
async def medium_operation():
    await asyncio.sleep(2)
    return "completed"