Add lint rule to ensure async functions are named X_async (#1326)

add lint rule to ensure async functions are named X_async

.flake8

@@ -9,7 +9,7 @@ per-file-ignores =
     */__init__.py: IMP100, E402, F401
     # we test various import patterns
     tests/test_exports.py: IMP100, IMP101, IMP102
-    tests/*: IMP101, IMP102, BAN100
+    tests/*: IMP101, IMP102, BAN100, ASY100
     # backcompat with outdated import patterns
     stripe/api_resources/*: IMP100, E402, F401
 
@@ -37,4 +37,5 @@ extension =
     SPY = flake8_stripe:TypingImportsChecker
     IMP = flake8_stripe:StripeImportsChecker
     BAN = flake8_stripe:BanPublicMethodsChecker
+    ASY = flake8_stripe:AsyncNamingConventions
 paths=./flake8_stripe

flake8_stripe/flake8_stripe.py

@@ -224,3 +224,30 @@ def run(self) -> Iterator[Tuple[int, int, str, type]]:
                         msg,
                         type(self),
                     )
+
+
+class AsyncNamingConventions:
+    name = __name__
+    version = "0.1.0"
+
+    def __init__(self, tree: ast.AST, filename: str):
+        self.tree = tree
+        self.filename = filename
+
+    def run(self) -> Iterator[Tuple[int, int, str, type]]:
+        for node in ast.walk(self.tree):
+            # ignore anything that isn't an async function declaration
+            if not isinstance(node, ast.AsyncFunctionDef):
+                continue
+
+            # dunders need specific names, so don't worry about them
+            if node.name.startswith("__") and node.name.endswith("__"):
+                continue
+
+            if not node.name.endswith("_async"):
+                yield (
+                    node.lineno,
+                    node.col_offset,
+                    "ASY100 Async methods must be named X_async",
+                    type(self),
+                )

stripe/_stripe_response.py

@@ -61,5 +61,6 @@ def __init__(
     def stream(self) -> AsyncIterable[bytes]:
         return self._stream
 
-    async def read(self) -> bytes:
+    # TODO (MAJOR): rename this to `read_async`
+    async def read(self) -> bytes:  # noqa: ASY100
         return b"".join([chunk async for chunk in self._stream])