11"""Calling functions with time limits."""
2- import multiprocessing
32import signal
4- from collections .abc import Callable , Generator , Iterable , Mapping
3+ from collections .abc import Generator
54from contextlib import contextmanager
6- from typing import Any , TypeVar
5+ from typing import TypeVar
76
87_T = TypeVar ("_T" )
98_V = TypeVar ("_V" )
109
11- __all__ = ("timed" , "time_limit" )
12-
13-
14- def timed (
15- func : Callable [[_T ], _V ],
16- args : Iterable = (),
17- kwds : Mapping [str , Any ] | None = None ,
18- timeout : float | None = None ,
19- ) -> _V :
20- """
21- Call a function with a time limit.
22-
23- Args:
24- func: Function to call.
25- args: Arguments for function.
26- kwds: Keyword arguments for function.
27- timeout: Timeout limit in seconds.
28-
29- Raises:
30- TimeoutError: If the function call takes longer than `timeout` seconds.
31- """
32- if kwds is None :
33- kwds = {}
34- with multiprocessing .Pool (1 , maxtasksperchild = 1 ) as pool :
35- result = pool .apply_async (func , args , kwds )
36- try :
37- return result .get (timeout )
38- except multiprocessing .TimeoutError as e :
39- raise TimeoutError (f"Call to { func .__name__ } timed out after { timeout } seconds." ) from e
10+ __all__ = ("time_limit" ,)
4011
4112
4213@contextmanager
@@ -51,7 +22,7 @@ def time_limit(timeout: int | None = None) -> Generator[None, None, None]:
5122 TimeoutError: If the function call takes longer than `timeout` seconds.
5223 """
5324
54- def signal_handler (signum , frame ):
25+ def signal_handler (_signum , _frame ):
5526 raise TimeoutError (f"time_limit call timed out after { timeout } seconds." )
5627
5728 signal .signal (signal .SIGALRM , signal_handler )
0 commit comments