diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 888978676..bfef4317f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -52,8 +52,10 @@ jobs: path: ~/.cache/pip key: pip-pyright - name: Install dependencies - run: pip install -e . pyright + run: pip install --group test -e .[trio] pyright - name: Run pyright + run: pyright src + - name: Run pyright --verifytypes run: pyright --ignoreexternal --verifytypes anyio test: diff --git a/src/anyio/__init__.py b/src/anyio/__init__.py index 2502c760b..fdbf1675f 100644 --- a/src/anyio/__init__.py +++ b/src/anyio/__init__.py @@ -92,11 +92,11 @@ from ._core._typedattr import typed_attribute as typed_attribute # Re-export imports so they look like they live directly in this package +__value = None for __value in list(locals().values()): if getattr(__value, "__module__", "").startswith("anyio."): __value.__module__ = __name__ - del __value diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index bc7e6ab38..63f7b2cab 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -177,9 +177,10 @@ def close(self) -> None: def get_loop(self) -> AbstractEventLoop: """Return embedded event loop.""" self._lazy_init() + assert self._loop is not None return self._loop - def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval: + def run(self, coro: Coroutine[Any, Any, T_Retval], *, context=None) -> T_Retval: """Run a coroutine inside the embedded event loop.""" if not coroutines.iscoroutine(coro): raise ValueError(f"a coroutine was expected, got {coro!r}") @@ -191,9 +192,11 @@ def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval: ) self._lazy_init() + assert self._loop is not None if context is None: context = self._context + assert context is not None task = context.run(self._loop.create_task, coro) if ( @@ -251,6 +254,7 @@ def _on_sigint(self, signum, frame, main_task: asyncio.Task) -> None: if self._interrupt_count == 1 and not main_task.done(): main_task.cancel() # wakeup loop if it is blocked by select() with long timeout + assert self._loop is not None self._loop.call_soon_threadsafe(lambda: None) return raise KeyboardInterrupt() @@ -287,8 +291,8 @@ def _do_shutdown(future: asyncio.futures.Future) -> None: except Exception as ex: loop.call_soon_threadsafe(future.set_exception, ex) - loop._executor_shutdown_called = True - if loop._default_executor is None: + loop._executor_shutdown_called = True # type: ignore[attr-defined] + if loop._default_executor is None: # type: ignore[attr-defined] return future = loop.create_future() thread = threading.Thread(target=_do_shutdown, args=(future,)) @@ -352,7 +356,10 @@ def _task_started(task: asyncio.Task) -> bool: # task list coro = task.get_coro() assert coro is not None - return getcoroutinestate(coro) in (CORO_RUNNING, CORO_SUSPENDED) + return getcoroutinestate(cast("Coroutine[Any, Any, object]", coro)) in ( + CORO_RUNNING, + CORO_SUSPENDED, + ) # @@ -497,7 +504,7 @@ def __exit__( if self._cancel_called and not self._parent_cancellation_is_visible_to_us: # For each level-cancel() call made on the host task, call uncancel() while self._pending_uncancellations: - self._host_task.uncancel() + self._host_task.uncancel() # type: ignore[attr-defined] self._pending_uncancellations -= 1 # Update cancelled_caught and check for exceptions we must not swallow @@ -725,7 +732,7 @@ def __init__(self, future: asyncio.Future, parent_id: int): self._future = future self._parent_id = parent_id - def started(self, value: T_contra | None = None) -> None: + def started(self, value: object | None = None) -> None: try: self._future.set_result(value) except asyncio.InvalidStateError: @@ -746,7 +753,7 @@ def started(self, value: T_contra | None = None) -> None: class TaskGroup(abc.TaskGroup): def __init__(self) -> None: - self.cancel_scope: CancelScope = CancelScope() + self.cancel_scope: abc.CancelScope = CancelScope() self._entered = False self._exceptions: list[BaseException] = [] self._tasks: set[asyncio.Task] = set() @@ -867,7 +874,7 @@ def task_done(_task: asyncio.Task) -> None: if not isinstance(exc, CancelledError): self._exceptions.append(exc) - if not self.cancel_scope._effectively_cancelled: + if not self.cancel_scope._effectively_cancelled: # type: ignore[attr-defined] self.cancel_scope.cancel() else: task_status_future.set_exception(exc) @@ -879,7 +886,7 @@ def task_done(_task: asyncio.Task) -> None: if task_status_future: parent_id = id(current_task()) else: - parent_id = id(self.cancel_scope._host_task) + parent_id = id(cast(CancelScope, self.cancel_scope)._host_task) handle = TaskHandle(coro, name) loop = asyncio.get_running_loop() @@ -896,12 +903,16 @@ def task_done(_task: asyncio.Task) -> None: # Make the spawned task inherit the task group's cancel scope _task_states[task] = TaskState( - parent_id=parent_id, cancel_scope=self.cancel_scope + parent_id=parent_id, cancel_scope=cast(CancelScope, self.cancel_scope) ) - self.cancel_scope._tasks.add(task) - self._tasks.add(task) - if sys.version_info >= (3, 14) and self.cancel_scope._host_task is not None: - asyncio.future_add_to_awaited_by(task, self.cancel_scope._host_task) + cast(CancelScope, self.cancel_scope)._tasks.add(task) + if ( + sys.version_info >= (3, 14) + and cast(CancelScope, self.cancel_scope)._host_task is not None + ): + asyncio.future_add_to_awaited_by( + task, cast(CancelScope, self.cancel_scope)._host_task + ) task.add_done_callback(task_done) return handle @@ -916,7 +927,7 @@ def create_task( if not isinstance(coro, Coroutine): raise TypeError(f"expected a coroutine, got {coro.__class__.__qualname__}") - if not self._entered or not self.cancel_scope._active: + if not self._entered or not cast(CancelScope, self.cancel_scope)._active: coro.close() raise RuntimeError( "This task group is not active; no new tasks can be started." @@ -934,14 +945,16 @@ async def start( name: object = None, return_handle: Literal[False] | Literal[True] = False, ) -> Any: - if not self._entered or not self.cancel_scope._active: + if not self._entered or not cast(CancelScope, self.cancel_scope)._active: raise RuntimeError( "This task group is not active; no new tasks can be started." ) future: asyncio.Future = asyncio.Future() final_name = get_callable_name(func, name) - task_status = _AsyncioTaskStatus(future, id(self.cancel_scope._host_task)) + task_status = _AsyncioTaskStatus( + future, id(cast(CancelScope, self.cancel_scope)._host_task) + ) coro = call_for_coroutine(func, args, task_status=task_status) handle = self._spawn(coro, final_name, future) @@ -1579,6 +1592,7 @@ async def accept(self) -> abc.SocketStream: with self._accept_guard: await AsyncIOBackend.checkpoint() + client_sock: socket.socket with CancelScope() as self._accept_scope: try: client_sock, _addr = await self._loop.sock_accept(self._raw_socket) @@ -1596,9 +1610,11 @@ async def accept(self) -> abc.SocketStream: finally: self._accept_scope = None - client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + assert client_sock is not None # type: ignore[possibly-unbound] + client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # type: ignore[possibly-unbound] transport, protocol = await self._loop.connect_accepted_socket( - StreamProtocol, client_sock + StreamProtocol, + client_sock, # type: ignore[possibly-unbound] ) return SocketStream(transport, protocol) @@ -2575,13 +2591,13 @@ def create_capacity_limiter(cls, total_tokens: float) -> abc.CapacityLimiter: return CapacityLimiter(total_tokens) @classmethod - async def run_sync_in_worker_thread( # type: ignore[return] + async def run_sync_in_worker_thread( cls, func: Callable[[Unpack[PosArgsT]], T_Retval], args: tuple[Unpack[PosArgsT]], abandon_on_cancel: bool = False, limiter: abc.CapacityLimiter | None = None, - ) -> T_Retval: + ) -> T_Retval: # type: ignore[return-value] await cls.checkpoint() # If this is the first run in this event loop thread, set up the necessary @@ -2656,6 +2672,7 @@ def run_async_from_thread( ) -> T_co: async def task_wrapper() -> T_co: __tracebackhide__ = True + task: asyncio.Task[object] if scope is not None: task = cast(asyncio.Task, current_task()) _task_states[task] = TaskState(None, scope) @@ -2666,7 +2683,7 @@ async def task_wrapper() -> T_co: raise concurrent.futures.CancelledError(str(exc)) from None finally: if scope is not None: - scope._tasks.discard(task) + scope._tasks.discard(task) # type: ignore[possibly-unbound] loop = cast( "AbstractEventLoop", token or threadlocals.current_token.native_token diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py index 341ddea19..a939273d4 100644 --- a/src/anyio/_backends/_trio.py +++ b/src/anyio/_backends/_trio.py @@ -161,10 +161,10 @@ def __exit__( exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> bool: - return self.__original.__exit__(exc_type, exc_val, exc_tb) + return cast(bool, self.__original.__exit__(exc_type, exc_val, exc_tb)) def cancel(self, reason: str | None = None) -> None: - self.__original.cancel(reason) + self.__original.cancel(reason) # type: ignore[call-arg] @property def deadline(self) -> float: @@ -200,7 +200,7 @@ def shield(self, value: bool) -> None: class _TrioTaskStatus(Generic[T_contra], abc.TaskStatus[T_contra]): early_start_value: T_contra | object = empty_start_value - real_task_status: trio.TaskStatus[T_contra | None] | None = None + real_task_status: trio.TaskStatus[T_contra | None] | None = None # type: ignore[name-defined] def started(self, value: T_contra | None = None) -> None: if self.real_task_status is None: @@ -240,7 +240,7 @@ async def __aexit__( return await self._nursery_manager.__aexit__(exc_type, exc_val, exc_tb) # type: ignore[return-value] except BaseExceptionGroup as exc: if not exc.split(trio.Cancelled)[1]: - raise trio.Cancelled._create() from exc + raise trio.Cancelled._create() from exc # type: ignore[attr-defined] raise finally: @@ -284,10 +284,11 @@ async def start( name: object = None, return_handle: Literal[False] | Literal[True] = False, ) -> Any: - handle: TaskHandle[T_co] + handle = TaskHandle(cast("Coroutine[Any, Any, T_co]", None), name) # type: ignore[arg-type] async def run_coro_with_task_status( - *, task_status: trio.TaskStatus[Any] + *, + task_status: trio.TaskStatus[Any], # type: ignore[name-defined] ) -> None: nonlocal handle wrapper_task_status = _TrioTaskStatus() @@ -303,11 +304,12 @@ async def run_coro_with_task_status( self._check_active() final_name = get_callable_name(func, name) start_value = await self._nursery.start( - run_coro_with_task_status, name=final_name + run_coro_with_task_status, + name=final_name, # type: ignore[arg-type] ) if return_handle: handle._start_value = start_value - return handle + return handle # type: ignore[return-value] else: return start_value @@ -567,7 +569,7 @@ async def send_fds(self, message: bytes, fds: Collection[int | IOBase]) -> None: ( socket.SOL_SOCKET, socket.SCM_RIGHTS, - fdarray, + bytes(fdarray), ) ], ) @@ -753,7 +755,7 @@ async def acquire(self) -> None: try: self.__original.acquire_nowait() except trio.WouldBlock: - await self.__original._lot.park() + await self.__original._lot.park() # type: ignore[attr-defined] except RuntimeError as exc: self._convert_runtime_error_msg(exc) raise @@ -813,7 +815,7 @@ async def acquire(self) -> None: try: self.__original.acquire_nowait() except trio.WouldBlock: - await self.__original._lot.park() + await self.__original._lot.park() # type: ignore[attr-defined] def acquire_nowait(self) -> None: try: @@ -977,7 +979,7 @@ def __exit__( self._call_queue.get()() def is_running(self) -> bool: - return trio.lowlevel.in_trio_task() + return trio.lowlevel.in_trio_task() # type: ignore[attr-defined] async def _run_tests_and_fixtures(self) -> None: self._send_stream, receive_stream = create_memory_object_stream[ @@ -1159,13 +1161,13 @@ def wrapper() -> T_Retval: token = TrioBackend.current_token() return await run_sync( wrapper, - abandon_on_cancel=abandon_on_cancel, + abandon_on_cancel=abandon_on_cancel, # type: ignore[call-arg] limiter=cast(trio.CapacityLimiter, limiter), ) @classmethod def check_cancelled(cls) -> None: - trio.from_thread.check_cancelled() + trio.from_thread.check_cancelled() # type: ignore[attr-defined] @classmethod def run_async_from_thread( @@ -1351,7 +1353,7 @@ async def getnameinfo( @classmethod async def wait_readable(cls, obj: FileDescriptorLike) -> None: try: - await wait_readable(obj) + await wait_readable(obj) # type: ignore[arg-type] except trio.ClosedResourceError as exc: raise ClosedResourceError().with_traceback(exc.__traceback__) from None except trio.BusyResourceError: @@ -1360,7 +1362,7 @@ async def wait_readable(cls, obj: FileDescriptorLike) -> None: @classmethod async def wait_writable(cls, obj: FileDescriptorLike) -> None: try: - await wait_writable(obj) + await wait_writable(obj) # type: ignore[arg-type] except trio.ClosedResourceError as exc: raise ClosedResourceError().with_traceback(exc.__traceback__) from None except trio.BusyResourceError: @@ -1368,7 +1370,7 @@ async def wait_writable(cls, obj: FileDescriptorLike) -> None: @classmethod def notify_closing(cls, obj: FileDescriptorLike) -> None: - notify_closing(obj) + notify_closing(obj) # type: ignore[arg-type] @classmethod async def wrap_listener_socket(cls, sock: socket.socket) -> abc.SocketListener: diff --git a/src/anyio/_core/_contextmanagers.py b/src/anyio/_core/_contextmanagers.py index 302f32b0c..e3f579849 100644 --- a/src/anyio/_core/_contextmanagers.py +++ b/src/anyio/_core/_contextmanagers.py @@ -67,7 +67,7 @@ def __enter__(self: _SupportsCtxMgr[_T_co, bool | None]) -> _T_co: f"'yield' statement?" ) - value = cm.__enter__() + value = cast(_T_co, cm.__enter__()) self.__cm = cm return value @@ -160,7 +160,7 @@ async def __aenter__(self: _SupportsAsyncCtxMgr[_T_co, bool | None]) -> _T_co: f"'yield' statement?" ) - value = await cm.__aenter__() + value = cast(_T_co, await cm.__aenter__()) self.__cm = cm return value diff --git a/src/anyio/_core/_sockets.py b/src/anyio/_core/_sockets.py index c75791b85..00842ed88 100644 --- a/src/anyio/_core/_sockets.py +++ b/src/anyio/_core/_sockets.py @@ -658,7 +658,13 @@ async def getaddrinfo( encoded_host, port, family=family, type=type, proto=proto, flags=flags ) return [ - (family, type, proto, canonname, convert_ipv6_sockaddr(sockaddr)) + ( + family, + type, + proto, + canonname, + convert_ipv6_sockaddr(cast("IPSockAddrType", sockaddr)), + ) for family, type, proto, canonname, sockaddr in gai_res # filter out IPv6 results when IPv6 is disabled if not isinstance(sockaddr[0], int) diff --git a/src/anyio/_core/_tempfile.py b/src/anyio/_core/_tempfile.py index 75a09f793..61916e2ce 100644 --- a/src/anyio/_core/_tempfile.py +++ b/src/anyio/_core/_tempfile.py @@ -327,7 +327,7 @@ async def rollover(self) -> None: self._fp = await to_thread.run_sync( lambda: tempfile.TemporaryFile(**self._tempfile_params) ) - await self.write(buffer.read()) + await self.write(buffer.read()) # type: ignore[arg-type] buffer.close() @property @@ -346,7 +346,7 @@ async def read1(self: SpooledTemporaryFile[bytes], size: int = -1) -> bytes: await checkpoint_if_cancelled() return self._fp.read1(size) - return await super().read1(size) + return await super().read1(size) # type: ignore[misc] async def readline(self) -> AnyStr: if not self._rolled: @@ -367,14 +367,14 @@ async def readinto(self: SpooledTemporaryFile[bytes], b: WriteableBuffer) -> int await checkpoint_if_cancelled() self._fp.readinto(b) - return await super().readinto(b) + return await super().readinto(b) # type: ignore[misc] async def readinto1(self: SpooledTemporaryFile[bytes], b: WriteableBuffer) -> int: if not self._rolled: await checkpoint_if_cancelled() self._fp.readinto(b) - return await super().readinto1(b) + return await super().readinto1(b) # type: ignore[misc] async def seek(self, offset: int, whence: int | None = os.SEEK_SET) -> int: if not self._rolled: @@ -551,7 +551,13 @@ async def mkstemp( :return: A tuple containing the file descriptor and the file name. """ - return await to_thread.run_sync(tempfile.mkstemp, suffix, prefix, dir, text) + return await to_thread.run_sync( # type: ignore[return-value] + tempfile.mkstemp, # type: ignore[arg-type] + suffix, + prefix, + dir, + text, + ) @overload @@ -586,7 +592,12 @@ async def mkdtemp( :return: The path of the created temporary directory. """ - return await to_thread.run_sync(tempfile.mkdtemp, suffix, prefix, dir) + return await to_thread.run_sync( # type: ignore[return-value] + tempfile.mkdtemp, # type: ignore[arg-type] + suffix, + prefix, + dir, + ) async def gettempdir() -> str: diff --git a/src/anyio/abc/__init__.py b/src/anyio/abc/__init__.py index d560ce3f1..0c737b84a 100644 --- a/src/anyio/abc/__init__.py +++ b/src/anyio/abc/__init__.py @@ -51,6 +51,7 @@ from ..from_thread import BlockingPortal as BlockingPortal # Re-export imports so they look like they live directly in this package +__value = None for __value in list(locals().values()): if getattr(__value, "__module__", "").startswith("anyio.abc."): __value.__module__ = __name__ diff --git a/src/anyio/from_thread.py b/src/anyio/from_thread.py index 8c7914c2f..c37bb5bb4 100644 --- a/src/anyio/from_thread.py +++ b/src/anyio/from_thread.py @@ -255,6 +255,7 @@ def callback(f: Future[T_Retval]) -> None: scope.cancel, "the future was cancelled", token=self._token ) + scope: CancelScope try: retval_or_awaitable = func(*args, **kwargs) if isawaitable(retval_or_awaitable): @@ -275,7 +276,7 @@ def callback(f: Future[T_Retval]) -> None: raise else: if not future.cancelled(): - future.set_result(retval) + future.set_result(retval) # type: ignore[arg-type, possibly-unbound] finally: scope = None # type: ignore[assignment] diff --git a/src/anyio/itertools.py b/src/anyio/itertools.py index 7e5248e4b..410f43d85 100644 --- a/src/anyio/itertools.py +++ b/src/anyio/itertools.py @@ -42,7 +42,7 @@ from ._core._tasks import CancelScope from .lowlevel import cancel_shielded_checkpoint, checkpoint, checkpoint_if_cancelled -T = TypeVar("T") +T = TypeVar("T", bound=object) R = TypeVar("R") _tee_end = object() @@ -141,7 +141,7 @@ async def __anext__(self) -> T: async def _operator_add(x: T, y: T) -> T: - return operator.add(x, y) + return operator.add(x, y) # type: ignore[arg-type, return-value] async def accumulate( diff --git a/src/anyio/pytest_plugin.py b/src/anyio/pytest_plugin.py index 5c667597d..16c4e81db 100644 --- a/src/anyio/pytest_plugin.py +++ b/src/anyio/pytest_plugin.py @@ -165,7 +165,7 @@ def pytest_pycollect_makeitem( collector: pytest.Module | pytest.Class, name: str, obj: object ) -> None: if collector.istestfunction(obj, name): - inner_func = obj.hypothesis.inner_test if hasattr(obj, "hypothesis") else obj + inner_func = obj.hypothesis.inner_test if hasattr(obj, "hypothesis") else obj # type: ignore[attr-defined] if iscoroutinefunction(inner_func): anyio_auto_mode = collector.config.getini("anyio_mode") == "auto" marker = collector.get_closest_marker("anyio") @@ -202,12 +202,12 @@ def pytest_collection_finish(session: pytest.Session) -> None: marks=[], ) else: # pytest 7.x - callspec = CallSpec2( # type: ignore[call-arg] - funcargs={}, + callspec = CallSpec2( + funcargs={}, # type: ignore[call-arg] params={"anyio_backend": backend}, indices={"anyio_backend": param_index}, - arg2scope={"anyio_backend": Scope.Module}, - idlist=[backend], + arg2scope={"anyio_backend": Scope.Module}, # type: ignore[call-arg] + idlist=[backend], # type: ignore[call-arg] marks=[], ) @@ -223,7 +223,7 @@ def pytest_collection_finish(session: pytest.Session) -> None: name2fixturedefs=fi.name2fixturedefs, ) new_item = pytest.Function.from_parent( - item.parent, + item.parent, # type: ignore[arg-type] name=f"{item.originalname}[{backend}]", callspec=callspec, callobj=item.obj, diff --git a/src/anyio/streams/tls.py b/src/anyio/streams/tls.py index 282174c71..adfa471e5 100644 --- a/src/anyio/streams/tls.py +++ b/src/anyio/streams/tls.py @@ -61,14 +61,14 @@ class TLSAttribute(TypedAttributeSet): cipher: tuple[str, str, int] = typed_attribute() #: the peer certificate in dictionary form (see :meth:`ssl.SSLSocket.getpeercert` # for more information) - peer_certificate: None | (dict[str, str | _PCTRTTT | _PCTRTT]) = typed_attribute() + peer_certificate: object = typed_attribute() #: the peer certificate in binary form peer_certificate_binary: bytes | None = typed_attribute() #: ``True`` if this is the server side of the connection server_side: bool = typed_attribute() #: ciphers shared by the client during the TLS handshake (``None`` if this is the #: client side) - shared_ciphers: list[tuple[str, str, int]] | None = typed_attribute() + shared_ciphers: object = typed_attribute() #: the :class:`~ssl.SSLObject` used for encryption ssl_object: ssl.SSLObject = typed_attribute() #: ``True`` if this stream does (and expects) a closing TLS handshake when the diff --git a/src/anyio/to_process.py b/src/anyio/to_process.py index 8d356fbd2..05547df30 100644 --- a/src/anyio/to_process.py +++ b/src/anyio/to_process.py @@ -42,12 +42,12 @@ _default_process_limiter: RunVar[CapacityLimiter] = RunVar("_default_process_limiter") -async def run_sync( # type: ignore[return] +async def run_sync( func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT], cancellable: bool = False, limiter: CapacityLimiter | None = None, -) -> T_Retval: +) -> T_Retval: # type: ignore[return-value] """ Call the given function with the given arguments in a worker process.