Add ASYNC233 pathlib blocking call rule

docs/changelog.rst

@@ -6,6 +6,7 @@ Changelog
 
 26.6.1
 ======
+- Add :ref:`ASYNC233 <async233>` blocking-pathlib-call to detect blocking ``pathlib.Path`` I/O methods in async functions. `(issue #396) <https://github.com/python-trio/flake8-async/issues/396>`_
 - Add :ref:`ASYNC127 <async127>` unmaintained-httpx: use ``httpx2`` instead of ``httpx``, which is no longer maintained, to get security updates. `(issue #460) <https://github.com/python-trio/flake8-async/issues/460>`_
 - :ref:`ASYNC210 <async210>`, :ref:`ASYNC211 <async211>` and :ref:`ASYNC212 <async212>` now also detect blocking calls made through ``httpx2``, and their messages recommend ``httpx2.AsyncClient`` instead of ``httpx.AsyncClient``.
 

docs/rules.rst

@@ -190,6 +190,9 @@ ASYNC231 : blocking-fdopen-call
 ASYNC232 : blocking-file-call
     Blocking sync call on file object, wrap the file object in :func:`trio.wrap_file`/:func:`anyio.wrap_file` to get an async file object.
 
+_`ASYNC233` : blocking-pathlib-call
+    Blocking sync call to ``pathlib.Path`` I/O methods in async function, use :class:`trio.Path`/:class:`anyio.Path`. ``asyncio`` users should consider `aiopath <https://pypi.org/project/aiopath>`__ or `anyio`_.
+
 ASYNC240 : blocking-path-usage
     Avoid using :mod:`os.path` in async functions, prefer using :class:`trio.Path`/:class:`anyio.Path` objects. ``asyncio`` users should consider `aiopath <https://pypi.org/project/aiopath>`__ or `anyio`_.
 

flake8_async/visitors/visitor2xx.py

@@ -4,7 +4,8 @@
 210 looks for usage of HTTP requests from common http libraries.
 211 additionally matches on object methods whose signature looks like an http request.
 220&221 looks for subprocess and os calls that should be wrapped.
-230&231 looks for os.open and os.fdopen that should be wrapped.
+230&231 looks for open and os.fdopen that should be wrapped.
+233 looks for pathlib.Path methods that should use an async path object.
 240 looks for os.path functions that interact with the disk in various ways.
 250 looks for input() that should be wrapped
 """
@@ -252,6 +253,7 @@ class Visitor23X(Visitor200):
     error_codes: Mapping[str, str] = {
         "ASYNC230": "Sync call {0} in async function, use `{1}.open_file(...)`.",
         "ASYNC231": "Sync call {0} in async function, use `{1}.wrap_file({0})`.",
+        "ASYNC233": "Sync call {0} on pathlib.Path in async function, use `{1}.Path`.",
         "ASYNC230_asyncio": (
             "Sync call {0} in async function, "
             " use a library such as aiofiles or anyio."
@@ -260,14 +262,104 @@ class Visitor23X(Visitor200):
             "Sync call {0} in async function, "
             " use a library such as aiofiles or anyio."
         ),
+        "ASYNC233_asyncio": (
+            "Sync call {0} on pathlib.Path in async function, "
+            " use `asyncio.loop.run_in_executor` or a library such as aiopath "
+            "or anyio."
+        ),
     }
 
+    pathlib_path_types = ("pathlib.Path", "pathlib.PosixPath", "pathlib.WindowsPath")
+    pathlib_path_constructors = pathlib_path_types + tuple(
+        f"{path_type}.{method}"
+        for path_type in pathlib_path_types
+        for method in ("cwd", "home")
+    )
+    pathlib_blocking_methods = (
+        "open",
+        "read_bytes",
+        "read_text",
+        "touch",
+        "write_bytes",
+        "write_text",
+    )
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self.pathlib_variables: set[str] = set()
+
+    def _is_pathlib_annotation(self, node: ast.AST | None) -> bool:
+        def or_none(node: ast.AST | None):
+            if not isinstance(node, ast.BinOp) or not isinstance(node.op, ast.BitOr):
+                return None
+            if isinstance(node.left, ast.Constant) and node.left.value is None:
+                return node.right
+            if isinstance(node.right, ast.Constant) and node.right.value is None:
+                return node.left
+            return None
+
+        if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name):
+            if node.value.id == "Optional":
+                node = node.slice
+        elif res := or_none(node):
+            node = res
+
+        return (
+            isinstance(node, (ast.Name, ast.Attribute))
+            and self.canonical_name(node) in self.pathlib_path_types
+        )
+
+    def visit_AsyncFunctionDef(
+        self, node: ast.AsyncFunctionDef | ast.FunctionDef | ast.Lambda
+    ):
+        self.save_state(node, "async_function", "pathlib_variables", copy=True)
+        self.async_function = isinstance(node, ast.AsyncFunctionDef)
+
+        args = node.args
+        for arg in *args.args, *args.posonlyargs, *args.kwonlyargs:
+            if self._is_pathlib_annotation(arg.annotation):
+                self.pathlib_variables.add(arg.arg)
+
+    visit_FunctionDef = visit_AsyncFunctionDef
+    visit_Lambda = visit_AsyncFunctionDef
+
+    def visit_ClassDef(self, node: ast.ClassDef):
+        self.save_state(node, "pathlib_variables", copy=True)
+
+    def visit_AnnAssign(self, node: ast.AnnAssign):
+        if isinstance(node.target, ast.Name) and self._is_pathlib_annotation(
+            node.annotation
+        ):
+            self.pathlib_variables.add(node.target.id)
+
+    def visit_Assign(self, node: ast.Assign):
+        if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name):
+            return
+        if self._is_pathlib_path_expr(node.value) or (
+            isinstance(node.value, ast.Name) and node.value.id in self.pathlib_variables
+        ):
+            self.pathlib_variables.add(node.targets[0].id)
+
     def visit_Call(self, node: ast.Call):
         canonical = self.canonical_name(node.func)
         if canonical in ("trio.wrap_file", "anyio.wrap_file") and len(node.args) == 1:
             setattr(node.args[0], "wrapped", True)  # noqa: B010
         super().visit_Call(node)
 
+    def _is_pathlib_path_expr(self, node: ast.expr) -> bool:
+        if isinstance(node, ast.Name):
+            return node.id in self.pathlib_variables
+        if isinstance(node, ast.Call):
+            return self.canonical_name(node.func) in self.pathlib_path_constructors
+        return False
+
+    def _is_pathlib_blocking_call(self, node: ast.Call) -> bool:
+        return (
+            isinstance(node.func, ast.Attribute)
+            and node.func.attr in self.pathlib_blocking_methods
+            and self._is_pathlib_path_expr(node.func.value)
+        )
+
     def visit_blocking_call(self, node: ast.Call):
         if getattr(node, "wrapped", False):
             return
@@ -277,6 +369,8 @@ def visit_blocking_call(self, node: ast.Call):
             error_code = "ASYNC230"
         elif canonical == "os.fdopen":
             error_code = "ASYNC231"
+        elif self._is_pathlib_blocking_call(node):
+            error_code = "ASYNC233"
         else:
             return
         if self.library == ("asyncio",):

tests/eval_files/async233.py

@@ -0,0 +1,35 @@
+# ARG --enable=ASYNC233
+# NOASYNCIO # see async233_asyncio.py
+import pathlib
+from pathlib import Path
+from pathlib import PosixPath as UnixPath
+
+import trio
+
+
+async def foo(path: Path, other_path: pathlib.Path):
+    path.open()  # ASYNC233: 4, 'path.open', "trio"
+    path.read_text()  # ASYNC233: 4, 'path.read_text', "trio"
+    path.read_bytes()  # ASYNC233: 4, 'path.read_bytes', "trio"
+    path.write_text("content")  # ASYNC233: 4, 'path.write_text', "trio"
+    path.write_bytes(b"content")  # ASYNC233: 4, 'path.write_bytes', "trio"
+    path.touch()  # ASYNC233: 4, 'path.touch', "trio"
+
+    other_path.read_text()  # ASYNC233: 4, 'other_path.read_text', "trio"
+    Path("foo").read_text()  # ASYNC233: 4, "Path('foo').read_text", "trio"
+    module_path = pathlib.Path("foo")
+    module_path.read_bytes()  # ASYNC233: 4, 'module_path.read_bytes', "trio"
+    cwd = pathlib.Path.cwd()
+    cwd.write_text("content")  # ASYNC233: 4, 'cwd.write_text', "trio"
+    UnixPath("foo").touch()  # ASYNC233: 4, "UnixPath('foo').touch", "trio"
+
+    assigned = Path("foo")
+    assigned.write_bytes(b"content")  # ASYNC233: 4, 'assigned.write_bytes', "trio"
+
+    await path.read_text()
+    await trio.wrap_file(path.open())
+    path.with_suffix(".txt")
+
+
+def foo_sync(path: Path):
+    path.read_text()

tests/eval_files/async233_asyncio.py

@@ -0,0 +1,14 @@
+# ARG --enable=ASYNC233
+# NOTRIO # see async233.py
+# NOANYIO # see async233.py
+# BASE_LIBRARY asyncio
+from pathlib import Path
+
+
+async def foo(path: Path):
+    path.read_text()  # ASYNC233_asyncio: 4, 'path.read_text'
+    Path("foo").write_text("content")  # ASYNC233_asyncio: 4, "Path('foo').write_text"
+
+
+def foo_sync(path: Path):
+    path.read_text()