[flake8-async] Implement blocking-path-method (ASYNC240).

Adds a new rule to find and report use of `os.path` or `pathlib.Path` in
async functions.

Author: PieterCK

crates/ruff_linter/resources/test/fixtures/flake8_async/ASYNC240.py

@@ -0,0 +1,87 @@
+import os
+from typing import Optional
+
+## Various os.path cases:
+
+def os_path_in_foo():
+    file = "file.txt"
+
+    os.path.abspath(file) # OK
+    os.path.exists(file) # OK
+
+async def os_path_in_foo():
+    file = "file.txt"
+
+    os.path.abspath(file) # ASYNC240
+    os.path.exists(file) # ASYNC240
+
+## Various pathlib.Path cases:
+from pathlib import Path
+
+def pathlib_path_in_foo():
+    path = Path("src/my_text.txt") # OK
+    path.absolute() # OK
+    path.exists() # OK
+    with path.open() as f: # OK
+        ...
+
+async def pathlib_path_in_foo():
+    path = Path("src/my_text.txt") # ASYNC240
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+    with path.open() as f: # ASYNC240
+        ...
+
+async def pathlib_path_in_foo():
+    import pathlib
+
+    path = pathlib.Path("src/my_text.txt") # ASYNC240
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+
+async def pathlib_path_in_fo():
+    from pathlib import Path as PathAlias
+
+    path = PathAlias("src/my_text.txt") # ASYNC240
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+
+global_path = Path("src/my_text.txt")
+
+async def pathlib_path_as_variable_in_foo():
+    global_path.absolute() # ASYNC240
+    global_path.exists() # ASYNC240
+
+def pathlib_path_in_foo(path: Path):
+    path.absolute() # OK
+    path.exists() # OK
+
+async def pathlib_path_in_foo(path: Path):
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+
+async def pathlib_path_in_foo(path: Path | None):
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+
+async def pathlib_path_in_foo(path: Optional[Path]):
+    path.absolute() # ASYNC240
+    path.exists() # ASYNC240
+
+## Valid cases using trio/anyio:
+
+async def trio_path_in_foo():
+    from trio import Path # OK
+
+    path = Path("src/my_text.txt") # OK
+    path.absolute() # OK
+    path.exists() # OK
+
+async def anyio_path_in_foo():
+    from anyio import Path # OK
+
+    path = Path("src/my_text.txt") # OK
+    path.absolute() # OK
+    path.exists() # OK
+
+

crates/ruff_linter/src/checkers/ast/analyze/expression.rs

@@ -666,6 +666,9 @@ pub(crate) fn expression(expr: &Expr, checker: &Checker) {
             if checker.is_rule_enabled(Rule::BlockingOpenCallInAsyncFunction) {
                 flake8_async::rules::blocking_open_call(checker, call);
             }
+            if checker.is_rule_enabled(Rule::BlockingPathMethodInAsyncFunction) {
+                flake8_async::rules::blocking_os_path(checker, call);
+            }
             if checker.any_rule_enabled(&[
                 Rule::CreateSubprocessInAsyncFunction,
                 Rule::RunProcessInAsyncFunction,

crates/ruff_linter/src/codes.rs

@@ -341,6 +341,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         (Flake8Async, "221") => (RuleGroup::Stable, rules::flake8_async::rules::RunProcessInAsyncFunction),
         (Flake8Async, "222") => (RuleGroup::Stable, rules::flake8_async::rules::WaitForProcessInAsyncFunction),
         (Flake8Async, "230") => (RuleGroup::Stable, rules::flake8_async::rules::BlockingOpenCallInAsyncFunction),
+        (Flake8Async, "240") => (RuleGroup::Preview, rules::flake8_async::rules::BlockingPathMethodInAsyncFunction),
         (Flake8Async, "250") => (RuleGroup::Preview, rules::flake8_async::rules::BlockingInputInAsyncFunction),
         (Flake8Async, "251") => (RuleGroup::Stable, rules::flake8_async::rules::BlockingSleepInAsyncFunction),
 

crates/ruff_linter/src/rules/flake8_async/mod.rs

@@ -28,6 +28,7 @@ mod tests {
     #[test_case(Rule::RunProcessInAsyncFunction, Path::new("ASYNC22x.py"))]
     #[test_case(Rule::WaitForProcessInAsyncFunction, Path::new("ASYNC22x.py"))]
     #[test_case(Rule::BlockingOpenCallInAsyncFunction, Path::new("ASYNC230.py"))]
+    #[test_case(Rule::BlockingPathMethodInAsyncFunction, Path::new("ASYNC240.py"))]
     #[test_case(Rule::BlockingInputInAsyncFunction, Path::new("ASYNC250.py"))]
     #[test_case(Rule::BlockingSleepInAsyncFunction, Path::new("ASYNC251.py"))]
     fn rules(rule_code: Rule, path: &Path) -> Result<()> {

crates/ruff_linter/src/rules/flake8_async/rules/blocking_path_methods.rs

@@ -0,0 +1,136 @@
+use crate::Violation;
+use crate::checkers::ast::Checker;
+use ruff_macros::{ViolationMetadata, derive_message_formats};
+use ruff_python_ast::{self as ast, Expr, ExprCall};
+use ruff_python_semantic::analyze::typing::{TypeChecker, check_type, traverse_union_and_optional};
+use ruff_text_size::Ranged;
+
+/// ## What it does
+/// Checks that async functions do not use `os.path`.
+///
+/// ## Why is this bad?
+/// Blocking an async function via `os.path` calls will block the entire
+/// event loop, preventing it from executing other tasks while waiting for the
+/// `os.path` operation, negating the benefits of asynchronous programming.
+///
+/// Instead of `os.path`, use `trio.Path` or `anyio.path` objects.
+///
+/// ## Example
+/// ```python
+/// import os
+///
+///
+/// async def func():
+///     path = "my_file.txt"
+///     file_exists = os.path.exists(path)
+/// ```
+///
+/// Use instead:
+/// ```python
+/// import trio
+///
+///
+/// async def func():
+///     path = trio.Path("my_file.txt")
+///     file_exists = await path.exists():
+/// ```
+#[derive(ViolationMetadata)]
+pub(crate) struct BlockingPathMethodInAsyncFunction;
+
+impl Violation for BlockingPathMethodInAsyncFunction {
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        "Async functions should not use pathlib.Path or os.path methods, use trio.Path or anyio.path".to_string()
+    }
+}
+
+const OS_PATH: [&str; 2] = ["os", "path"];
+const PATHLIB_PATH: [&str; 2] = ["pathlib", "Path"];
+
+fn is_blocking_path_method(segments: &[&str]) -> bool {
+    segments.starts_with(&OS_PATH) || segments.starts_with(&PATHLIB_PATH)
+}
+
+struct PathMethodInAsyncChecker;
+
+impl TypeChecker for PathMethodInAsyncChecker {
+    fn match_annotation(
+        annotation: &ruff_python_ast::Expr,
+        semantic: &ruff_python_semantic::SemanticModel,
+    ) -> bool {
+        // match base annotation directly
+        if semantic
+            .resolve_qualified_name(annotation)
+            .is_some_and(|qualified_name| is_blocking_path_method(qualified_name.segments()))
+        {
+            return true;
+        }
+
+        // otherwise traverse any union or optional annotation
+        let mut found = false;
+        traverse_union_and_optional(
+            &mut |inner_expr, _| {
+                if semantic
+                    .resolve_qualified_name(inner_expr)
+                    .is_some_and(|qualified_name| {
+                        is_blocking_path_method(qualified_name.segments())
+                    })
+                {
+                    found = true;
+                }
+            },
+            semantic,
+            annotation,
+        );
+        found
+    }
+
+    fn match_initializer(
+        initializer: &ruff_python_ast::Expr,
+        semantic: &ruff_python_semantic::SemanticModel,
+    ) -> bool {
+        let Expr::Call(ExprCall { func, .. }) = initializer else {
+            return false;
+        };
+
+        semantic
+            .resolve_qualified_name(func)
+            .is_some_and(|qualified_name| is_blocking_path_method(qualified_name.segments()))
+    }
+}
+
+/// ASYNC240
+pub(crate) fn blocking_os_path(checker: &Checker, call: &ExprCall) {
+    let semantic = checker.semantic();
+    if !semantic.in_async_context() {
+        return;
+    }
+
+    // Check if the expression is calling the blocking path method
+    // using its imported symbol/name.
+    if let Some(qualified_name) = semantic.resolve_qualified_name(call.func.as_ref()) {
+        let segments = qualified_name.segments();
+        if is_blocking_path_method(segments){
+            checker.report_diagnostic(BlockingPathMethodInAsyncFunction, call.func.range());
+        }
+        return;
+    }
+
+    // Past this, we're checking if the expression is binded to one
+    // of the blocking path methods.
+    let Some(ast::ExprAttribute { value, .. }) = call.func.as_attribute_expr() else {
+        return;
+    };
+
+    let Some(name) = value.as_name_expr() else {
+        return;
+    };
+
+    let Some(binding) = semantic.only_binding(name).map(|id| semantic.binding(id)) else {
+        return;
+    };
+
+    if check_type::<PathMethodInAsyncChecker>(binding, semantic) {
+        checker.report_diagnostic(BlockingPathMethodInAsyncFunction, call.func.range());
+    }
+}

crates/ruff_linter/src/rules/flake8_async/rules/mod.rs

@@ -5,6 +5,7 @@ pub(crate) use blocking_http_call::*;
 pub(crate) use blocking_http_call_httpx::*;
 pub(crate) use blocking_input::*;
 pub(crate) use blocking_open_call::*;
+pub(crate) use blocking_path_methods::*;
 pub(crate) use blocking_process_invocation::*;
 pub(crate) use blocking_sleep::*;
 pub(crate) use cancel_scope_no_checkpoint::*;
@@ -18,6 +19,7 @@ mod blocking_http_call;
 mod blocking_http_call_httpx;
 mod blocking_input;
 mod blocking_open_call;
+mod blocking_path_methods;
 mod blocking_process_invocation;
 mod blocking_sleep;
 mod cancel_scope_no_checkpoint;