Add type stubs from typeshed

.github/workflows/main.yml

@@ -32,6 +32,10 @@ jobs:
         if: matrix.python-version == '3.13' && runner.os == 'Linux'
         run: "uv run --group tox tox -e lint"
 
+      - name: "Typecheck"
+        if: matrix.python-version == '3.13' && runner.os == 'Linux'
+        run: "uv run --group tox tox -e typecheck"
+
       - name: "Upload coverage data"
         uses: "actions/upload-artifact@v4"
         with:

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 25.1.0 (UNRELEASED)
 
+- Added type stubs from [typeshed](https://github.com/python/typeshed)
 - Switch to [uv](https://docs.astral.sh/uv/) + add Python v3.14 support.
   ([#219](https://github.com/Tinche/aiofiles/pull/219))
 - Add `ruff` formatter and linter.

Justfile

@@ -15,5 +15,8 @@ format:
 lint: format
 	{{ run_prefix }}ruff check --fix {{ code_dirs }}
 
+typecheck:
+	{{ run_prefix }}mypy src
+
 test:
-	{{ run_prefix }}pytest -x --ff {{ tests_dir }}
+	{{ run_prefix }}pytest -x --ff {{ tests_dir }}

README.md

@@ -181,3 +181,8 @@ async def test_stuff():
 
 Contributions are very welcome. Tests can be run with `tox`, please ensure
 the coverage at least stays the same before you submit a pull request.
+
+If your change touches the public API or any function/class signatures, please
+also update the type stub files (`*.pyi`) to match the runtime code.  Please
+also run a type check locally (e.g., `tox -e typecheck`) before you submit a
+pull request

pyproject.toml

@@ -136,7 +136,7 @@ fixable = [
 
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
-"src/**/*.py" = [
+"src/**/*.{py,pyi}" = [
     "TID252",  # https://docs.astral.sh/ruff/rules/relative-imports/
 ]
 "tests/**/*.py" = [

src/aiofiles/__init__.pyi

@@ -0,0 +1,33 @@
+from . import tempfile as tempfile
+from .threadpool import (
+    open as open,
+)
+from .threadpool import (
+    stderr as stderr,
+)
+from .threadpool import (
+    stderr_bytes as stderr_bytes,
+)
+from .threadpool import (
+    stdin as stdin,
+)
+from .threadpool import (
+    stdin_bytes as stdin_bytes,
+)
+from .threadpool import (
+    stdout as stdout,
+)
+from .threadpool import (
+    stdout_bytes as stdout_bytes,
+)
+
+__all__ = [
+    "open",
+    "tempfile",
+    "stdin",
+    "stdout",
+    "stderr",
+    "stdin_bytes",
+    "stdout_bytes",
+    "stderr_bytes",
+]

src/aiofiles/base.pyi

@@ -0,0 +1,41 @@
+from asyncio.events import AbstractEventLoop
+from collections.abc import Awaitable, Callable, Generator
+from concurrent.futures import Executor
+from contextlib import AbstractAsyncContextManager
+from types import TracebackType
+from typing import Any, BinaryIO, Generic, TextIO, TypeVar
+
+from typing_extensions import Self
+
+_T = TypeVar("_T")
+_V_co = TypeVar("_V_co", covariant=True)
+
+class AsyncBase(Generic[_T]):
+    def __init__(
+        self,
+        file: TextIO | BinaryIO | None,
+        loop: AbstractEventLoop | None,
+        executor: Executor | None,
+    ) -> None: ...
+    def __aiter__(self) -> Self: ...
+    async def __anext__(self) -> _T: ...
+
+class AsyncIndirectBase(AsyncBase[_T]):
+    def __init__(
+        self,
+        name: str,
+        loop: AbstractEventLoop | None,
+        executor: Executor | None,
+        indirect: Callable[[], TextIO | BinaryIO],
+    ) -> None: ...
+
+class AiofilesContextManager(Awaitable[_V_co], AbstractAsyncContextManager[_V_co]):
+    def __init__(self, coro: Awaitable[_V_co]) -> None: ...
+    def __await__(self) -> Generator[Any, Any, _V_co]: ...
+    async def __aenter__(self) -> _V_co: ...
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None: ...

src/aiofiles/os.pyi

@@ -0,0 +1,219 @@
+import sys
+from asyncio.events import AbstractEventLoop
+from collections.abc import Sequence
+from concurrent.futures import Executor
+from os import _ScandirIterator, stat_result
+from typing import AnyStr, overload
+
+from _typeshed import (
+    BytesPath,
+    FileDescriptorOrPath,
+    GenericPath,
+    ReadableBuffer,
+    StrOrBytesPath,
+    StrPath,
+)
+
+from aiofiles import ospath
+from aiofiles.ospath import wrap as wrap
+
+__all__ = [
+    "path",
+    "stat",
+    "rename",
+    "renames",
+    "replace",
+    "remove",
+    "unlink",
+    "mkdir",
+    "makedirs",
+    "rmdir",
+    "removedirs",
+    "link",
+    "symlink",
+    "readlink",
+    "listdir",
+    "scandir",
+    "access",
+    "wrap",
+    "getcwd",
+]
+
+path = ospath
+
+async def stat(
+    path: FileDescriptorOrPath,
+    *,
+    dir_fd: int | None = None,
+    follow_symlinks: bool = True,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> stat_result: ...
+async def rename(
+    src: StrOrBytesPath,
+    dst: StrOrBytesPath,
+    *,
+    src_dir_fd: int | None = None,
+    dst_dir_fd: int | None = None,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def renames(
+    old: StrOrBytesPath,
+    new: StrOrBytesPath,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def replace(
+    src: StrOrBytesPath,
+    dst: StrOrBytesPath,
+    *,
+    src_dir_fd: int | None = None,
+    dst_dir_fd: int | None = None,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def remove(
+    path: StrOrBytesPath,
+    *,
+    dir_fd: int | None = None,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def unlink(
+    path: StrOrBytesPath,
+    *,
+    dir_fd: int | None = ...,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def mkdir(
+    path: StrOrBytesPath,
+    mode: int = 511,
+    *,
+    dir_fd: int | None = None,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def makedirs(
+    name: StrOrBytesPath,
+    mode: int = 511,
+    exist_ok: bool = False,
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def link(
+    src: StrOrBytesPath,
+    dst: StrOrBytesPath,
+    *,
+    src_dir_fd: int | None = ...,
+    dst_dir_fd: int | None = ...,
+    follow_symlinks: bool = ...,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def symlink(
+    src: StrOrBytesPath,
+    dst: StrOrBytesPath,
+    target_is_directory: bool = ...,
+    *,
+    dir_fd: int | None = ...,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def readlink(
+    path: AnyStr,
+    *,
+    dir_fd: int | None = ...,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> AnyStr: ...
+async def rmdir(
+    path: StrOrBytesPath,
+    *,
+    dir_fd: int | None = None,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+async def removedirs(
+    name: StrOrBytesPath,
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> None: ...
+@overload
+async def scandir(
+    path: None = None,
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> _ScandirIterator[str]: ...
+@overload
+async def scandir(
+    path: int, *, loop: AbstractEventLoop | None = ..., executor: Executor | None = ...
+) -> _ScandirIterator[str]: ...
+@overload
+async def scandir(
+    path: GenericPath[AnyStr],
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> _ScandirIterator[AnyStr]: ...
+@overload
+async def listdir(
+    path: StrPath | None,
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> list[str]: ...
+@overload
+async def listdir(
+    path: BytesPath,
+    *,
+    loop: AbstractEventLoop | None = ...,
+    executor: Executor | None = ...,
+) -> list[bytes]: ...
+@overload
+async def listdir(
+    path: int, *, loop: AbstractEventLoop | None = ..., executor: Executor | None = ...
+) -> list[str]: ...
+async def access(
+    path: FileDescriptorOrPath,
+    mode: int,
+    *,
+    dir_fd: int | None = None,
+    effective_ids: bool = False,
+    follow_symlinks: bool = True,
+) -> bool: ...
+async def getcwd() -> str: ...
+
+if sys.platform != "win32":
+    from os import statvfs_result
+
+    @overload
+    async def sendfile(
+        out_fd: int,
+        in_fd: int,
+        offset: int | None,
+        count: int,
+        *,
+        loop: AbstractEventLoop | None = ...,
+        executor: Executor | None = ...,
+    ) -> int: ...
+    @overload
+    async def sendfile(
+        out_fd: int,
+        in_fd: int,
+        offset: int,
+        count: int,
+        headers: Sequence[ReadableBuffer] = ...,
+        trailers: Sequence[ReadableBuffer] = ...,
+        flags: int = ...,
+        *,
+        loop: AbstractEventLoop | None = ...,
+        executor: Executor | None = ...,
+    ) -> int: ...  # FreeBSD and Mac OS X only
+    async def statvfs(path: FileDescriptorOrPath) -> statvfs_result: ...  # Unix only
+
+    __all__ += ["statvfs", "sendfile"]