Repository: apify/crawlee-python
File: src/crawlee/_utils/time.py (Python)

Code before:
_SECONDS_PER_HOUR = 3600


@dataclass
class TimerResult:
    wall: float | None = None
    cpu: float | None = None


@contextmanager
def measure_time() -> Iterator[TimerResult]:
    """Measure the execution time (wall-clock and CPU) between the start and end of the with-block."""
    result = TimerResult()
    before_wall = time.monotonic()
    before_cpu = time.thread_time()

    try:
        yield result
    finally:
        after_wall = time.monotonic()
        after_cpu = time.thread_time()
        result.wall = after_wall - before_wall
        result.cpu = after_cpu - before_cpu


class SharedTimeout:
    """Keeps track of a time budget shared by multiple independent async operations."""

    def __init__(self, timeout: timedelta) -> None:
        self._remaining_timeout = timeout
        self._active_timeout: Timeout | None = None
        self._activation_timestamp: float | None = None

    async def __aenter__(self) -> timedelta:
        if self._active_timeout is not None or self._activation_timestamp is not None:
            raise RuntimeError('A shared timeout context cannot be entered twice at the same time')

        self._activation_timestamp = time.monotonic()
        self._active_timeout = new_timeout = timeout(self._remaining_timeout.total_seconds())
        await new_timeout.__aenter__()
        return self._remaining_timeout

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        if self._active_timeout is None or self._activation_timestamp is None:
            raise RuntimeError('Logic error')


Review comment (suggestion):
Nice idea. I like this a lot. 

Just thinking out loud, where this could lead to:
In case we need to create more granular timeouts for specific steps, I think this class could be easily expanded to support that. I am wondering if even the final context consumer (request handler) could just use timeout from here if the timeout is set on the context (my other comment)








Code after:
_SECONDS_PER_HOUR = 3600


@dataclass
class TimerResult:
    wall: float | None = None
    cpu: float | None = None


@contextmanager
def measure_time() -> Iterator[TimerResult]:
    """Measure the execution time (wall-clock and CPU) between the start and end of the with-block."""
    result = TimerResult()
    before_wall = time.monotonic()
    before_cpu = time.thread_time()

    try:
        yield result
    finally:
        after_wall = time.monotonic()
        after_cpu = time.thread_time()
        result.wall = after_wall - before_wall
        result.cpu = after_cpu - before_cpu


class SharedTimeout:
    """Keeps track of a time budget shared by multiple independent async operations.

    Provides a reusable, non-reentrant context manager interface.
    """

    def __init__(self, timeout: timedelta) -> None:
        self._remaining_timeout = timeout
        self._active_timeout: Timeout | None = None
        self._activation_timestamp: float | None = None

    async def __aenter__(self) -> timedelta:
        if self._active_timeout is not None or self._activation_timestamp is not None:
            raise RuntimeError('A shared timeout context cannot be entered twice at the same time')

        self._activation_timestamp = time.monotonic()
        self._active_timeout = new_timeout = timeout(self._remaining_timeout.total_seconds())
        await new_timeout.__aenter__()
        return self._remaining_timeout

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None: