Add type annotations for wait_for2

asyncua/common/utils.py

@@ -141,6 +141,11 @@ def fields_with_resolved_types(
 async def wait_for(aw: Awaitable[_T], timeout: Union[int, float, None]) -> _T:
     """
     Wrapped version of asyncio.wait_for that does not swallow cancellations
+
+    There is a bug in asyncio.wait_for before Python version 3.12 that prevents the inner awaitable from being cancelled
+    when the task is cancelled from the outside.
+
+    See https://github.com/python/cpython/issues/87555 and https://github.com/python/cpython/issues/86296
     """
     if sys.version_info >= (3, 12):
         return await asyncio.wait_for(aw, timeout)

setup.cfg

@@ -11,6 +11,7 @@ max-line-length = 160
 disable_error_code = misc, arg-type, assignment, var-annotated
 show_error_codes = True
 check_untyped_defs = False
+mypy_path = ./stubs
 [mypy-asyncua.ua.uaprotocol_auto.*]
 # Autogenerated file
 disable_error_code = literal-required

setup.py

@@ -20,7 +20,7 @@
     author="Olivier Roulet-Dubonnet",
     author_email="[email protected]",
     url='http://freeopcua.github.io/',
-    packages=find_packages(exclude=["tests"]),
+    packages=find_packages(exclude=["tests", "stubs"]),
     provides=["asyncua"],
     license="GNU Lesser General Public License v3 or later",
     install_requires=["aiofiles", "aiosqlite", "python-dateutil", "pytz", "cryptography>42.0.0", "sortedcontainers", "importlib-metadata;python_version<'3.8'", "pyOpenSSL>23.2.0", "typing-extensions", 'wait_for2==0.3.2'],

stubs/wait_for2/__init__.pyi

@@ -0,0 +1,14 @@
+import asyncio
+from typing import Any, Awaitable, Callable, TypeVar, Union
+
+_T = TypeVar('_T')
+
+
+async def wait_for(
+    fut: Awaitable[_T],
+    timeout: Union[int, float, None],
+    *,
+    loop: asyncio.AbstractEventLoop = None,
+    race_handler: Callable[[Union[_T, BaseException], bool], Any] = None,
+):
+    ...