SONARPY-3002 Rule S7515: async with should be used for asynchronous resource management (#318)

GitOrigin-RevId: a9fb1fed2a8d5a4bcddf47850fc56553f3616dca

Author: guillaume-dequenne

python-checks/src/main/java/org/sonar/python/checks/AsyncWithContextManagerCheck.java

@@ -0,0 +1,78 @@
+/*
+ * SonarQube Python Plugin
+ * Copyright (C) 2011-2025 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the Sonar Source-Available License Version 1, as published by SonarSource SA.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the Sonar Source-Available License for more details.
+ *
+ * You should have received a copy of the Sonar Source-Available License
+ * along with this program; if not, see https://sonarsource.com/license/ssal/
+ */
+package org.sonar.python.checks;
+
+import org.sonar.check.Rule;
+import org.sonar.plugins.python.api.PythonSubscriptionCheck;
+import org.sonar.plugins.python.api.SubscriptionContext;
+import org.sonar.plugins.python.api.tree.Expression;
+import org.sonar.plugins.python.api.tree.Tree;
+import org.sonar.plugins.python.api.tree.WithItem;
+import org.sonar.plugins.python.api.tree.WithStatement;
+import org.sonar.plugins.python.api.types.v2.PythonType;
+import org.sonar.plugins.python.api.types.v2.TriBool;
+import org.sonar.python.tree.TreeUtils;
+import org.sonar.python.types.v2.TypeCheckBuilder;
+
+@Rule(key = "S7515")
+public class AsyncWithContextManagerCheck extends PythonSubscriptionCheck {
+
+  private static final String MESSAGE = "Use \"async with\" instead of \"with\" for this asynchronous context manager.";
+  private static final String SECONDARY_TYPE_DEFINITION = "This context manager implements the async context manager protocol.";
+  private static final String SECONDARY_MESSAGE = "This function is async.";
+
+  private TypeCheckBuilder hasAsyncEnter;
+  private TypeCheckBuilder hasAsyncExit;
+
+  @Override
+  public void initialize(Context context) {
+    context.registerSyntaxNodeConsumer(Tree.Kind.FILE_INPUT, this::setupTypeChecker);
+    context.registerSyntaxNodeConsumer(Tree.Kind.WITH_STMT, this::checkWithStatement);
+  }
+
+  private void setupTypeChecker(SubscriptionContext ctx) {
+    hasAsyncEnter = ctx.typeChecker().typeCheckBuilder().hasMember("__aenter__");
+    hasAsyncExit = ctx.typeChecker().typeCheckBuilder().hasMember("__aexit__");
+  }
+
+  private void checkWithStatement(SubscriptionContext ctx) {
+    WithStatement withStatement = (WithStatement) ctx.syntaxNode();
+    if (withStatement.isAsync()) {
+      return;
+    }
+    var asyncToken = TreeUtils.asyncTokenOfEnclosingFunction(withStatement).orElse(null);
+    if (asyncToken == null) {
+      return;
+    }
+    for (WithItem item : withStatement.withItems()) {
+      Expression contextManager = item.test();
+      PythonType contextManagerType = contextManager.typeV2();
+      if (implementsAsyncContextManagerProtocol(contextManagerType)) {
+        PreciseIssue issue = ctx.addIssue(withStatement.withKeyword(), MESSAGE)
+          .secondary(asyncToken, SECONDARY_MESSAGE);
+        contextManagerType.definitionLocation().ifPresent(
+          location -> issue.secondary(location, SECONDARY_TYPE_DEFINITION)
+        );
+        break;
+      }
+    }
+  }
+
+  private boolean implementsAsyncContextManagerProtocol(PythonType type) {
+    return hasAsyncEnter.check(type) == TriBool.TRUE && hasAsyncExit.check(type) == TriBool.TRUE;
+  }
+}

python-checks/src/main/java/org/sonar/python/checks/OpenSourceCheckList.java

@@ -125,6 +125,7 @@ public Stream<Class<?>> getChecks() {
       AsyncFunctionWithTimeoutCheck.class,
       AsyncioTaskNotStoredCheck.class,
       AsyncLongSleepCheck.class,
+      AsyncWithContextManagerCheck.class,
       BackslashInStringCheck.class,
       BackticksUsageCheck.class,
       BareRaiseInFinallyCheck.class,

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/S7515.html

@@ -0,0 +1,76 @@
+<p>This rule raises when using a regular <code>with</code> statement inside an async function with a context manager that implements the asynchronous
+context manager protocol.</p>
+<h2>Why is this an issue?</h2>
+<p>When working within an async function, it is important to maintain consistency with the asynchronous programming model. If a context manager
+implements the asynchronous context manager protocol (defining <code>__aenter__</code> and <code>__aexit__</code> methods), it should be used with the
+<code>async with</code> statement rather than the regular <code>with</code> statement.</p>
+<p>The asynchronous context manager protocol is specifically designed to handle resources that may require asynchronous setup or teardown operations.
+Using the regular <code>with</code> statement in an async context bypasses this intended asynchronous behavior.</p>
+<h3>What is the potential impact?</h3>
+<p>Not following the proper async pattern can lead to:</p>
+<ul>
+  <li> <strong>Inconsistent async usage</strong>: Mixing synchronous and asynchronous patterns reduces code clarity </li>
+  <li> <strong>Missed async opportunities</strong>: Asynchronous setup and cleanup operations may not be utilized </li>
+  <li> <strong>Maintenance issues</strong>: Future developers may not understand the intended async behavior </li>
+</ul>
+<h2>How to fix it</h2>
+<p>Use the <code>async with</code> statement when working with asynchronous context managers.</p>
+<h3>Code examples</h3>
+<h4>Noncompliant code example</h4>
+<pre data-diff-id="1" data-diff-type="noncompliant">
+class Resource:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        ...
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        ...
+
+async def main():
+    resource = Resource()
+    with resource:  # Noncompliant: using 'with' in async function when async protocol is available
+        ...
+</pre>
+<h4>Compliant solution</h4>
+<pre data-diff-id="1" data-diff-type="compliant">
+class Resource:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        ...
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        ...
+
+async def main():
+    async with Resource() as resource:  # Compliant: using 'async with' in async function
+        ...
+</pre>
+<h3>How does this work?</h3>
+<p>The <code>async with</code> statement provides the proper way to use asynchronous context managers:</p>
+<ol>
+  <li> It calls the <code>__aenter__</code> method and awaits its result </li>
+  <li> Assigns the returned value to the variable after <code>as</code> (if specified) </li>
+  <li> Executes the code block within the context </li>
+  <li> Calls the <code>__aexit__</code> method and awaits its completion, even if an exception occurs </li>
+</ol>
+<p>This ensures consistency with the async programming model and allows the context manager to perform any necessary asynchronous operations during
+setup and cleanup.</p>
+<h2>Resources</h2>
+<h3>Documentation</h3>
+<ul>
+  <li> Python Documentation - <a href="https://docs.python.org/3/reference/datamodel.html#asynchronous-context-managers">Asynchronous Context
+  Managers</a> </li>
+  <li> Python Documentation - <a href="https://docs.python.org/3/reference/compound_stmts.html#the-async-with-statement">The async with statement</a>
+  </li>
+</ul>
+

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/S7515.json

@@ -0,0 +1,24 @@
+{
+  "title": "\"async with\" should be used for asynchronous resource management",
+  "type": "BUG",
+  "status": "ready",
+  "remediation": {
+    "func": "Constant\/Issue",
+    "constantCost": "5min"
+  },
+  "tags": [
+    "async"
+  ],
+  "defaultSeverity": "Major",
+  "ruleSpecification": "RSPEC-7515",
+  "sqKey": "S7515",
+  "scope": "All",
+  "quickfix": "targeted",
+  "code": {
+    "impacts": {
+      "MAINTAINABILITY": "LOW",
+      "RELIABILITY": "LOW"
+    },
+    "attribute": "LOGICAL"
+  }
+}

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/Sonar_way_profile.json

@@ -283,6 +283,7 @@
     "S7512",
     "S7513",
     "S7514",
+    "S7515",
     "S7516",
     "S7517",
     "S7519"

python-checks/src/test/java/org/sonar/python/checks/AsyncWithContextManagerCheckTest.java

@@ -0,0 +1,28 @@
+/*
+ * SonarQube Python Plugin
+ * Copyright (C) 2011-2025 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the Sonar Source-Available License Version 1, as published by SonarSource SA.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the Sonar Source-Available License for more details.
+ *
+ * You should have received a copy of the Sonar Source-Available License
+ * along with this program; if not, see https://sonarsource.com/license/ssal/
+ */
+package org.sonar.python.checks;
+
+import org.junit.jupiter.api.Test;
+import org.sonar.python.checks.utils.PythonCheckVerifier;
+
+class AsyncWithContextManagerCheckTest {
+
+  @Test
+  void test() {
+    PythonCheckVerifier.verify("src/test/resources/checks/asyncWithContextManager.py", new AsyncWithContextManagerCheck());
+  }
+}

python-checks/src/test/resources/checks/asyncWithContextManager.py

@@ -0,0 +1,88 @@
+# Test cases for async context managers
+
+class AsyncContextManager:
+#     ^^^^^^^^^^^^^^^^^^^> {{This context manager implements the async context manager protocol.}}
+    async def __aenter__(self):
+        print("Entering async context")
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        print("Exiting async context")
+        return False
+
+class RegularContextManager:
+    def __enter__(self):
+        print("Entering regular context")
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        print("Exiting regular context")
+        return False
+
+class DualContextManager:
+    # Implements both sync and async protocols
+    def __enter__(self):
+        print("Entering regular context")
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        print("Exiting regular context")
+        return False
+
+    async def __aenter__(self):
+        print("Entering async context")
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        print("Exiting async context")
+        return False
+
+async def async_function_with_async_context_manager():
+#^[sc=1;ec=5]> {{This function is async.}}
+    with AsyncContextManager() as acm:  # Noncompliant {{Use "async with" instead of "with" for this asynchronous context manager.}}
+#   ^^^^
+        print("Inside async context")
+
+async def async_function_with_dual_context_manager():
+    with DualContextManager() as dcm:  # Noncompliant
+        print("Inside context")
+
+# Compliant examples
+async def async_function_with_async_with():
+    async with AsyncContextManager() as acm:  # Compliant
+        print("Inside async context")
+
+async def async_function_with_regular_context_manager():
+    with RegularContextManager() as rcm:  # Compliant - no async protocol
+        print("Inside regular context")
+
+async def async_function_with_dual_context_manager_compliant():
+    async with DualContextManager() as dcm:  # Compliant
+        print("Inside async context")
+
+# Regular function - should be compliant regardless
+def regular_function():
+    with AsyncContextManager() as acm:  # Compliant - not in async function
+        print("Inside async context")
+    
+    with DualContextManager() as dcm:  # Compliant - not in async function
+        print("Inside context")
+
+# Nested contexts
+async def nested_contexts():
+    with RegularContextManager() as outer:  # Compliant - no async protocol
+        with AsyncContextManager() as inner:  # Noncompliant
+            print("Nested contexts")
+
+# Class method examples
+class MyClass:
+    async def async_method(self):
+        with AsyncContextManager() as acm:  # Noncompliant
+            print("Inside async method")
+        
+        async with AsyncContextManager() as acm:  # Compliant
+            print("Inside async method with async with")
+    
+    def regular_method(self):
+        with AsyncContextManager() as acm:  # Compliant - not in async method
+            print("Inside regular method")