async-timeout

from async_timeout import timeout, timeout_at

from async_timeout import timeout, timeout_at, Timeout

from async_timeout import timeout, timeout_at, Timeout
# Available exports: timeout, timeout_at, Timeout

from async_timeout import timeout
import asyncio
import aiohttp

async def main():
    # Basic relative timeout
    async with timeout(1.5):
        await some_async_operation()
    
    # Absolute timeout
    loop = asyncio.get_running_loop()
    async with timeout_at(loop.time() + 10):
        await some_async_operation()
    
    # Check if timeout occurred
    async with timeout(2.0) as cm:
        try:
            await some_async_operation()
        except asyncio.TimeoutError:
            if cm.expired:
                print("Operation timed out")

def timeout(delay: Optional[float]) -> Timeout:
    """
    Create a timeout context manager with relative delay.
    
    Args:
        delay: Value in seconds or None to disable timeout logic
    
    Returns:
        Timeout context manager instance
    """

def timeout_at(deadline: Optional[float]) -> Timeout:
    """
    Create a timeout context manager with absolute deadline.
    
    Args:
        deadline: Absolute time in loop.time() clock system or None
    
    Returns:
        Timeout context manager instance
    """

class Timeout:
    """
    Timeout context manager for asyncio programs.
    
    Use via async with statement for timeout functionality.
    Provides methods for checking expiration and modifying timeouts during execution.
    """
    
    @property
    def expired(self) -> bool:
        """
        Check if timeout expired during execution.
        
        Note:
            In Python 3.11+, this is also callable: expired() returns bool
        
        Returns:
            True if timeout occurred, False otherwise
        """
    
    @property
    def deadline(self) -> Optional[float]:
        """
        Get the deadline time in loop.time() clock system.
        
        Returns:
            Float timestamp or None if no deadline set
        """
    
    def reject(self) -> None:
        """
        Cancel/reject scheduled timeout.
        
        Disables the timeout while context manager is active.
        """
    
    def shift(self, delay: float) -> None:
        """
        Advance timeout by delay seconds (can be negative).
        
        Args:
            delay: Number of seconds to shift (positive or negative)
        
        Raises:
            RuntimeError: If deadline is not scheduled
        """
    
    def update(self, deadline: float) -> None:
        """
        Set absolute deadline value.
        
        Args:
            deadline: Absolute time in loop.time() clock system
        
        Note:
            If deadline is in past, timeout is raised immediately
        """
    
    def when(self) -> Optional[float]:
        """
        Get the scheduled deadline time (Python 3.11+ only).
        
        Returns:
            Float timestamp or None if no deadline set
        
        Note:
            This method is only available in Python 3.11+ as part of asyncio.Timeout compatibility
        """
    
    def reschedule(self, deadline: Optional[float]) -> None:
        """
        Reschedule the timeout to a new deadline (Python 3.11+ only).
        
        Args:
            deadline: New absolute deadline or None to cancel timeout
        
        Note:
            This method is only available in Python 3.11+ as part of asyncio.Timeout compatibility
        """

import asyncio
import aiohttp
from async_timeout import timeout

async def fetch_with_timeout():
    async with timeout(5.0):
        async with aiohttp.ClientSession() as session:
            async with session.get('https://example.com') as response:
                return await response.text()

# Usage
try:
    result = await fetch_with_timeout()
    print("Request completed successfully")
except asyncio.TimeoutError:
    print("Request timed out after 5 seconds")

import asyncio
from async_timeout import timeout

async def adaptive_timeout_example():
    async with timeout(3.0) as cm:
        # Start some operation
        await asyncio.sleep(1)
        
        # Check if we need more time and extend timeout
        if some_condition:
            cm.shift(2.0)  # Add 2 more seconds
            print(f"Extended deadline to: {cm.deadline}")
        
        # Continue operation
        await asyncio.sleep(2)
        
        # Check if timeout occurred
        if cm.expired:
            print("Operation was cancelled due to timeout")
        else:
            print("Operation completed within timeout")

import asyncio
from async_timeout import timeout

async def conditional_timeout():
    async with timeout(2.0) as cm:
        # Start operation
        await asyncio.sleep(0.5)
        
        # Conditionally disable timeout
        if fast_path_available:
            cm.reject()  # Remove timeout
            await fast_operation()
        else:
            await slow_operation()

import asyncio
from async_timeout import timeout_at

async def scheduled_timeout():
    loop = asyncio.get_running_loop()
    start_time = loop.time()
    
    # Set absolute deadline 10 seconds from now
    async with timeout_at(start_time + 10) as cm:
        await some_long_operation()
        
        # Update to new absolute deadline if needed
        if need_more_time:
            cm.update(start_time + 15)

import asyncio
import sys
from async_timeout import timeout

async def python311_features():
    if sys.version_info >= (3, 11):
        async with timeout(3.0) as cm:
            # Use when() method to get deadline (Python 3.11+ only)
            current_deadline = cm.when()
            print(f"Current deadline: {current_deadline}")
            
            # Use reschedule() method (Python 3.11+ only)
            cm.reschedule(current_deadline + 2.0)
            
            # expired can be called as method or accessed as property
            if cm.expired():  # Method call
                print("Timeout expired (method)")
            if cm.expired:    # Property access
                print("Timeout expired (property)")
            
            await some_operation()

import asyncio
from async_timeout import timeout

async def error_handling_example():
    try:
        async with timeout(1.0) as cm:
            await some_operation()
    except asyncio.TimeoutError:
        if cm.expired:
            print("Operation timed out")
        else:
            print("TimeoutError from operation itself")
    except RuntimeError as e:
        print(f"Invalid timeout operation: {e}")

import sys

if sys.version_info >= (3, 11):
    from asyncio import timeout, timeout_at
else:
    from async_timeout import timeout, timeout_at