We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
No module named 'exceptiongroup' when upgraded to starlette 0.31.0
After upgrading to starlette 0.31.0 today, we run into the following exception when importing starlette:

```
from starlette import applications, requests, responses, routing
2023-07-24 15:47:43    File "/opt/conda/lib/python3.8/site-packages/starlette/applications.py", line 8, in <module>
2023-07-24 15:47:43      from starlette.middleware.base import BaseHTTPMiddleware
2023-07-24 15:47:43    File "/opt/conda/lib/python3.8/site-packages/starlette/middleware/base.py", line 14, in <module>
2023-07-24 15:47:43      from exceptiongroup import BaseExceptionGroup
2023-07-24 15:47:43  ModuleNotFoundError: No module named 'exceptiongroup'
```

This is likely introduced from https://github.com/encode/starlette/pull/2211/files
Apply AnyIO upgrade changes only on test suite
- Closes https://github.com/encode/starlette/issues/2227
- Closes https://github.com/encode/starlette/discussions/2228

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `starlette/_utils.py`
Content:
```
1 import asyncio
2 import functools
3 import sys
4 import typing
5 
6 if sys.version_info >= (3, 10):  # pragma: no cover
7     from typing import TypeGuard
8 else:  # pragma: no cover
9     from typing_extensions import TypeGuard
10 
11 T = typing.TypeVar("T")
12 AwaitableCallable = typing.Callable[..., typing.Awaitable[T]]
13 
14 
15 @typing.overload
16 def is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]:
17     ...
18 
19 
20 @typing.overload
21 def is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]:
22     ...
23 
24 
25 def is_async_callable(obj: typing.Any) -> typing.Any:
26     while isinstance(obj, functools.partial):
27         obj = obj.func
28 
29     return asyncio.iscoroutinefunction(obj) or (
30         callable(obj) and asyncio.iscoroutinefunction(obj.__call__)
31     )
32 
33 
34 T_co = typing.TypeVar("T_co", covariant=True)
35 
36 
37 class AwaitableOrContextManager(
38     typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]
39 ):
40     ...
41 
42 
43 class SupportsAsyncClose(typing.Protocol):
44     async def close(self) -> None:
45         ...  # pragma: no cover
46 
47 
48 SupportsAsyncCloseType = typing.TypeVar(
49     "SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False
50 )
51 
52 
53 class AwaitableOrContextManagerWrapper(typing.Generic[SupportsAsyncCloseType]):
54     __slots__ = ("aw", "entered")
55 
56     def __init__(self, aw: typing.Awaitable[SupportsAsyncCloseType]) -> None:
57         self.aw = aw
58 
59     def __await__(self) -> typing.Generator[typing.Any, None, SupportsAsyncCloseType]:
60         return self.aw.__await__()
61 
62     async def __aenter__(self) -> SupportsAsyncCloseType:
63         self.entered = await self.aw
64         return self.entered
65 
66     async def __aexit__(self, *args: typing.Any) -> typing.Union[None, bool]:
67         await self.entered.close()
68         return None
69
```

Path: `starlette/middleware/base.py`
Content:
```
1 import sys
2 import typing
3 from contextlib import contextmanager
4 
5 import anyio
6 from anyio.abc import ObjectReceiveStream, ObjectSendStream
7 
8 from starlette.background import BackgroundTask
9 from starlette.requests import ClientDisconnect, Request
10 from starlette.responses import ContentStream, Response, StreamingResponse
11 from starlette.types import ASGIApp, Message, Receive, Scope, Send
12 
13 if sys.version_info < (3, 11):  # pragma: no cover
14     from exceptiongroup import BaseExceptionGroup
15 
16 RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
17 DispatchFunction = typing.Callable[
18     [Request, RequestResponseEndpoint], typing.Awaitable[Response]
19 ]
20 T = typing.TypeVar("T")
21 
22 
23 @contextmanager
24 def _convert_excgroups() -> typing.Generator[None, None, None]:
25     try:
26         yield
27     except BaseException as exc:
28         while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:
29             exc = exc.exceptions[0]
30 
31         raise exc
32 
33 
34 class _CachedRequest(Request):
35     """
36     If the user calls Request.body() from their dispatch function
37     we cache the entire request body in memory and pass that to downstream middlewares,
38     but if they call Request.stream() then all we do is send an
39     empty body so that downstream things don't hang forever.
40     """
41 
42     def __init__(self, scope: Scope, receive: Receive):
43         super().__init__(scope, receive)
44         self._wrapped_rcv_disconnected = False
45         self._wrapped_rcv_consumed = False
46         self._wrapped_rc_stream = self.stream()
47 
48     async def wrapped_receive(self) -> Message:
49         # wrapped_rcv state 1: disconnected
50         if self._wrapped_rcv_disconnected:
51             # we've already sent a disconnect to the downstream app
52             # we don't need to wait to get another one
53             # (although most ASGI servers will just keep sending it)
54             return {"type": "http.disconnect"}
55         # wrapped_rcv state 1: consumed but not yet disconnected
56         if self._wrapped_rcv_consumed:
57             # since the downstream app has consumed us all that is left
58             # is to send it a disconnect
59             if self._is_disconnected:
60                 # the middleware has already seen the disconnect
61                 # since we know the client is disconnected no need to wait
62                 # for the message
63                 self._wrapped_rcv_disconnected = True
64                 return {"type": "http.disconnect"}
65             # we don't know yet if the client is disconnected or not
66             # so we'll wait until we get that message
67             msg = await self.receive()
68             if msg["type"] != "http.disconnect":  # pragma: no cover
69                 # at this point a disconnect is all that we should be receiving
70                 # if we get something else, things went wrong somewhere
71                 raise RuntimeError(f"Unexpected message received: {msg['type']}")
72             return msg
73 
74         # wrapped_rcv state 3: not yet consumed
75         if getattr(self, "_body", None) is not None:
76             # body() was called, we return it even if the client disconnected
77             self._wrapped_rcv_consumed = True
78             return {
79                 "type": "http.request",
80                 "body": self._body,
81                 "more_body": False,
82             }
83         elif self._stream_consumed:
84             # stream() was called to completion
85             # return an empty body so that downstream apps don't hang
86             # waiting for a disconnect
87             self._wrapped_rcv_consumed = True
88             return {
89                 "type": "http.request",
90                 "body": b"",
91                 "more_body": False,
92             }
93         else:
94             # body() was never called and stream() wasn't consumed
95             try:
96                 stream = self.stream()
97                 chunk = await stream.__anext__()
98                 self._wrapped_rcv_consumed = self._stream_consumed
99                 return {
100                     "type": "http.request",
101                     "body": chunk,
102                     "more_body": not self._stream_consumed,
103                 }
104             except ClientDisconnect:
105                 self._wrapped_rcv_disconnected = True
106                 return {"type": "http.disconnect"}
107 
108 
109 class BaseHTTPMiddleware:
110     def __init__(
111         self, app: ASGIApp, dispatch: typing.Optional[DispatchFunction] = None
112     ) -> None:
113         self.app = app
114         self.dispatch_func = self.dispatch if dispatch is None else dispatch
115 
116     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
117         if scope["type"] != "http":
118             await self.app(scope, receive, send)
119             return
120 
121         request = _CachedRequest(scope, receive)
122         wrapped_receive = request.wrapped_receive
123         response_sent = anyio.Event()
124 
125         async def call_next(request: Request) -> Response:
126             app_exc: typing.Optional[Exception] = None
127             send_stream: ObjectSendStream[typing.MutableMapping[str, typing.Any]]
128             recv_stream: ObjectReceiveStream[typing.MutableMapping[str, typing.Any]]
129             send_stream, recv_stream = anyio.create_memory_object_stream()
130 
131             async def receive_or_disconnect() -> Message:
132                 if response_sent.is_set():
133                     return {"type": "http.disconnect"}
134 
135                 async with anyio.create_task_group() as task_group:
136 
137                     async def wrap(func: typing.Callable[[], typing.Awaitable[T]]) -> T:
138                         result = await func()
139                         task_group.cancel_scope.cancel()
140                         return result
141 
142                     task_group.start_soon(wrap, response_sent.wait)
143                     message = await wrap(wrapped_receive)
144 
145                 if response_sent.is_set():
146                     return {"type": "http.disconnect"}
147 
148                 return message
149 
150             async def close_recv_stream_on_response_sent() -> None:
151                 await response_sent.wait()
152                 recv_stream.close()
153 
154             async def send_no_error(message: Message) -> None:
155                 try:
156                     await send_stream.send(message)
157                 except anyio.BrokenResourceError:
158                     # recv_stream has been closed, i.e. response_sent has been set.
159                     return
160 
161             async def coro() -> None:
162                 nonlocal app_exc
163 
164                 async with send_stream:
165                     try:
166                         await self.app(scope, receive_or_disconnect, send_no_error)
167                     except Exception as exc:
168                         app_exc = exc
169 
170             task_group.start_soon(close_recv_stream_on_response_sent)
171             task_group.start_soon(coro)
172 
173             try:
174                 message = await recv_stream.receive()
175                 info = message.get("info", None)
176                 if message["type"] == "http.response.debug" and info is not None:
177                     message = await recv_stream.receive()
178             except anyio.EndOfStream:
179                 if app_exc is not None:
180                     raise app_exc
181                 raise RuntimeError("No response returned.")
182 
183             assert message["type"] == "http.response.start"
184 
185             async def body_stream() -> typing.AsyncGenerator[bytes, None]:
186                 async with recv_stream:
187                     async for message in recv_stream:
188                         assert message["type"] == "http.response.body"
189                         body = message.get("body", b"")
190                         if body:
191                             yield body
192                         if not message.get("more_body", False):
193                             break
194 
195                 if app_exc is not None:
196                     raise app_exc
197 
198             response = _StreamingResponse(
199                 status_code=message["status"], content=body_stream(), info=info
200             )
201             response.raw_headers = message["headers"]
202             return response
203 
204         with _convert_excgroups():
205             async with anyio.create_task_group() as task_group:
206                 response = await self.dispatch_func(request, call_next)
207                 await response(scope, wrapped_receive, send)
208                 response_sent.set()
209 
210     async def dispatch(
211         self, request: Request, call_next: RequestResponseEndpoint
212     ) -> Response:
213         raise NotImplementedError()  # pragma: no cover
214 
215 
216 class _StreamingResponse(StreamingResponse):
217     def __init__(
218         self,
219         content: ContentStream,
220         status_code: int = 200,
221         headers: typing.Optional[typing.Mapping[str, str]] = None,
222         media_type: typing.Optional[str] = None,
223         background: typing.Optional[BackgroundTask] = None,
224         info: typing.Optional[typing.Mapping[str, typing.Any]] = None,
225     ) -> None:
226         self._info = info
227         super().__init__(content, status_code, headers, media_type, background)
228 
229     async def stream_response(self, send: Send) -> None:
230         if self._info:
231             await send({"type": "http.response.debug", "info": self._info})
232         return await super().stream_response(send)
233
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/starlette/_utils.py b/starlette/_utils.py
--- a/starlette/_utils.py
+++ b/starlette/_utils.py
@@ -2,12 +2,20 @@
 import functools
 import sys
 import typing
+from contextlib import contextmanager
 
 if sys.version_info >= (3, 10):  # pragma: no cover
     from typing import TypeGuard
 else:  # pragma: no cover
     from typing_extensions import TypeGuard
 
+has_exceptiongroups = True
+if sys.version_info < (3, 11):  # pragma: no cover
+    try:
+        from exceptiongroup import BaseExceptionGroup
+    except ImportError:
+        has_exceptiongroups = False
+
 T = typing.TypeVar("T")
 AwaitableCallable = typing.Callable[..., typing.Awaitable[T]]
 
@@ -66,3 +74,15 @@
     async def __aexit__(self, *args: typing.Any) -> typing.Union[None, bool]:
         await self.entered.close()
         return None
+
+
+@contextmanager
+def collapse_excgroups() -> typing.Generator[None, None, None]:
+    try:
+        yield
+    except BaseException as exc:
+        if has_exceptiongroups:
+            while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:
+                exc = exc.exceptions[0]  # pragma: no cover
+
+        raise exc
diff --git a/starlette/middleware/base.py b/starlette/middleware/base.py
--- a/starlette/middleware/base.py
+++ b/starlette/middleware/base.py
@@ -1,18 +1,14 @@
-import sys
 import typing
-from contextlib import contextmanager
 
 import anyio
 from anyio.abc import ObjectReceiveStream, ObjectSendStream
 
+from starlette._utils import collapse_excgroups
 from starlette.background import BackgroundTask
 from starlette.requests import ClientDisconnect, Request
 from starlette.responses import ContentStream, Response, StreamingResponse
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
-if sys.version_info < (3, 11):  # pragma: no cover
-    from exceptiongroup import BaseExceptionGroup
-
 RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
 DispatchFunction = typing.Callable[
     [Request, RequestResponseEndpoint], typing.Awaitable[Response]
@@ -20,17 +16,6 @@
 T = typing.TypeVar("T")
 
 
-@contextmanager
-def _convert_excgroups() -> typing.Generator[None, None, None]:
-    try:
-        yield
-    except BaseException as exc:
-        while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:
-            exc = exc.exceptions[0]
-
-        raise exc
-
-
 class _CachedRequest(Request):
     """
     If the user calls Request.body() from their dispatch function
@@ -201,7 +186,7 @@
             response.raw_headers = message["headers"]
             return response
 
-        with _convert_excgroups():
+        with collapse_excgroups():
             async with anyio.create_task_group() as task_group:
                 response = await self.dispatch_func(request, call_next)
                 await response(scope, wrapped_receive, send)

{"golden_diff": "diff --git a/starlette/_utils.py b/starlette/_utils.py\n--- a/starlette/_utils.py\n+++ b/starlette/_utils.py\n@@ -2,12 +2,20 @@\n import functools\n import sys\n import typing\n+from contextlib import contextmanager\n \n if sys.version_info >= (3, 10):  # pragma: no cover\n     from typing import TypeGuard\n else:  # pragma: no cover\n     from typing_extensions import TypeGuard\n \n+has_exceptiongroups = True\n+if sys.version_info < (3, 11):  # pragma: no cover\n+    try:\n+        from exceptiongroup import BaseExceptionGroup\n+    except ImportError:\n+        has_exceptiongroups = False\n+\n T = typing.TypeVar(\"T\")\n AwaitableCallable = typing.Callable[..., typing.Awaitable[T]]\n \n@@ -66,3 +74,15 @@\n     async def __aexit__(self, *args: typing.Any) -> typing.Union[None, bool]:\n         await self.entered.close()\n         return None\n+\n+\n+@contextmanager\n+def collapse_excgroups() -> typing.Generator[None, None, None]:\n+    try:\n+        yield\n+    except BaseException as exc:\n+        if has_exceptiongroups:\n+            while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:\n+                exc = exc.exceptions[0]  # pragma: no cover\n+\n+        raise exc\ndiff --git a/starlette/middleware/base.py b/starlette/middleware/base.py\n--- a/starlette/middleware/base.py\n+++ b/starlette/middleware/base.py\n@@ -1,18 +1,14 @@\n-import sys\n import typing\n-from contextlib import contextmanager\n \n import anyio\n from anyio.abc import ObjectReceiveStream, ObjectSendStream\n \n+from starlette._utils import collapse_excgroups\n from starlette.background import BackgroundTask\n from starlette.requests import ClientDisconnect, Request\n from starlette.responses import ContentStream, Response, StreamingResponse\n from starlette.types import ASGIApp, Message, Receive, Scope, Send\n \n-if sys.version_info < (3, 11):  # pragma: no cover\n-    from exceptiongroup import BaseExceptionGroup\n-\n RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\n DispatchFunction = typing.Callable[\n     [Request, RequestResponseEndpoint], typing.Awaitable[Response]\n@@ -20,17 +16,6 @@\n T = typing.TypeVar(\"T\")\n \n \n-@contextmanager\n-def _convert_excgroups() -> typing.Generator[None, None, None]:\n-    try:\n-        yield\n-    except BaseException as exc:\n-        while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:\n-            exc = exc.exceptions[0]\n-\n-        raise exc\n-\n-\n class _CachedRequest(Request):\n     \"\"\"\n     If the user calls Request.body() from their dispatch function\n@@ -201,7 +186,7 @@\n             response.raw_headers = message[\"headers\"]\n             return response\n \n-        with _convert_excgroups():\n+        with collapse_excgroups():\n             async with anyio.create_task_group() as task_group:\n                 response = await self.dispatch_func(request, call_next)\n                 await response(scope, wrapped_receive, send)\n", "issue": "No module named 'exceptiongroup' when upgraded to starlette 0.31.0\nAfter upgrading to starlette 0.31.0 today, we run into the following exception when importing starlette:\r\n\r\n```\r\nfrom starlette import applications, requests, responses, routing\r\n2023-07-24 15:47:43    File \"/opt/conda/lib/python3.8/site-packages/starlette/applications.py\", line 8, in <module>\r\n2023-07-24 15:47:43      from starlette.middleware.base import BaseHTTPMiddleware\r\n2023-07-24 15:47:43    File \"/opt/conda/lib/python3.8/site-packages/starlette/middleware/base.py\", line 14, in <module>\r\n2023-07-24 15:47:43      from exceptiongroup import BaseExceptionGroup\r\n2023-07-24 15:47:43  ModuleNotFoundError: No module named 'exceptiongroup'\r\n```\r\n\r\nThis is likely introduced from https://github.com/encode/starlette/pull/2211/files\nApply AnyIO upgrade changes only on test suite\n- Closes https://github.com/encode/starlette/issues/2227\r\n- Closes https://github.com/encode/starlette/discussions/2228\n", "before_files": [{"content": "import asyncio\nimport functools\nimport sys\nimport typing\n\nif sys.version_info >= (3, 10):  # pragma: no cover\n    from typing import TypeGuard\nelse:  # pragma: no cover\n    from typing_extensions import TypeGuard\n\nT = typing.TypeVar(\"T\")\nAwaitableCallable = typing.Callable[..., typing.Awaitable[T]]\n\n\n@typing.overload\ndef is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]:\n    ...\n\n\n@typing.overload\ndef is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]:\n    ...\n\n\ndef is_async_callable(obj: typing.Any) -> typing.Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return asyncio.iscoroutinefunction(obj) or (\n        callable(obj) and asyncio.iscoroutinefunction(obj.__call__)\n    )\n\n\nT_co = typing.TypeVar(\"T_co\", covariant=True)\n\n\nclass AwaitableOrContextManager(\n    typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]\n):\n    ...\n\n\nclass SupportsAsyncClose(typing.Protocol):\n    async def close(self) -> None:\n        ...  # pragma: no cover\n\n\nSupportsAsyncCloseType = typing.TypeVar(\n    \"SupportsAsyncCloseType\", bound=SupportsAsyncClose, covariant=False\n)\n\n\nclass AwaitableOrContextManagerWrapper(typing.Generic[SupportsAsyncCloseType]):\n    __slots__ = (\"aw\", \"entered\")\n\n    def __init__(self, aw: typing.Awaitable[SupportsAsyncCloseType]) -> None:\n        self.aw = aw\n\n    def __await__(self) -> typing.Generator[typing.Any, None, SupportsAsyncCloseType]:\n        return self.aw.__await__()\n\n    async def __aenter__(self) -> SupportsAsyncCloseType:\n        self.entered = await self.aw\n        return self.entered\n\n    async def __aexit__(self, *args: typing.Any) -> typing.Union[None, bool]:\n        await self.entered.close()\n        return None\n", "path": "starlette/_utils.py"}, {"content": "import sys\nimport typing\nfrom contextlib import contextmanager\n\nimport anyio\nfrom anyio.abc import ObjectReceiveStream, ObjectSendStream\n\nfrom starlette.background import BackgroundTask\nfrom starlette.requests import ClientDisconnect, Request\nfrom starlette.responses import ContentStream, Response, StreamingResponse\nfrom starlette.types import ASGIApp, Message, Receive, Scope, Send\n\nif sys.version_info < (3, 11):  # pragma: no cover\n    from exceptiongroup import BaseExceptionGroup\n\nRequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\nDispatchFunction = typing.Callable[\n    [Request, RequestResponseEndpoint], typing.Awaitable[Response]\n]\nT = typing.TypeVar(\"T\")\n\n\n@contextmanager\ndef _convert_excgroups() -> typing.Generator[None, None, None]:\n    try:\n        yield\n    except BaseException as exc:\n        while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:\n            exc = exc.exceptions[0]\n\n        raise exc\n\n\nclass _CachedRequest(Request):\n    \"\"\"\n    If the user calls Request.body() from their dispatch function\n    we cache the entire request body in memory and pass that to downstream middlewares,\n    but if they call Request.stream() then all we do is send an\n    empty body so that downstream things don't hang forever.\n    \"\"\"\n\n    def __init__(self, scope: Scope, receive: Receive):\n        super().__init__(scope, receive)\n        self._wrapped_rcv_disconnected = False\n        self._wrapped_rcv_consumed = False\n        self._wrapped_rc_stream = self.stream()\n\n    async def wrapped_receive(self) -> Message:\n        # wrapped_rcv state 1: disconnected\n        if self._wrapped_rcv_disconnected:\n            # we've already sent a disconnect to the downstream app\n            # we don't need to wait to get another one\n            # (although most ASGI servers will just keep sending it)\n            return {\"type\": \"http.disconnect\"}\n        # wrapped_rcv state 1: consumed but not yet disconnected\n        if self._wrapped_rcv_consumed:\n            # since the downstream app has consumed us all that is left\n            # is to send it a disconnect\n            if self._is_disconnected:\n                # the middleware has already seen the disconnect\n                # since we know the client is disconnected no need to wait\n                # for the message\n                self._wrapped_rcv_disconnected = True\n                return {\"type\": \"http.disconnect\"}\n            # we don't know yet if the client is disconnected or not\n            # so we'll wait until we get that message\n            msg = await self.receive()\n            if msg[\"type\"] != \"http.disconnect\":  # pragma: no cover\n                # at this point a disconnect is all that we should be receiving\n                # if we get something else, things went wrong somewhere\n                raise RuntimeError(f\"Unexpected message received: {msg['type']}\")\n            return msg\n\n        # wrapped_rcv state 3: not yet consumed\n        if getattr(self, \"_body\", None) is not None:\n            # body() was called, we return it even if the client disconnected\n            self._wrapped_rcv_consumed = True\n            return {\n                \"type\": \"http.request\",\n                \"body\": self._body,\n                \"more_body\": False,\n            }\n        elif self._stream_consumed:\n            # stream() was called to completion\n            # return an empty body so that downstream apps don't hang\n            # waiting for a disconnect\n            self._wrapped_rcv_consumed = True\n            return {\n                \"type\": \"http.request\",\n                \"body\": b\"\",\n                \"more_body\": False,\n            }\n        else:\n            # body() was never called and stream() wasn't consumed\n            try:\n                stream = self.stream()\n                chunk = await stream.__anext__()\n                self._wrapped_rcv_consumed = self._stream_consumed\n                return {\n                    \"type\": \"http.request\",\n                    \"body\": chunk,\n                    \"more_body\": not self._stream_consumed,\n                }\n            except ClientDisconnect:\n                self._wrapped_rcv_disconnected = True\n                return {\"type\": \"http.disconnect\"}\n\n\nclass BaseHTTPMiddleware:\n    def __init__(\n        self, app: ASGIApp, dispatch: typing.Optional[DispatchFunction] = None\n    ) -> None:\n        self.app = app\n        self.dispatch_func = self.dispatch if dispatch is None else dispatch\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        if scope[\"type\"] != \"http\":\n            await self.app(scope, receive, send)\n            return\n\n        request = _CachedRequest(scope, receive)\n        wrapped_receive = request.wrapped_receive\n        response_sent = anyio.Event()\n\n        async def call_next(request: Request) -> Response:\n            app_exc: typing.Optional[Exception] = None\n            send_stream: ObjectSendStream[typing.MutableMapping[str, typing.Any]]\n            recv_stream: ObjectReceiveStream[typing.MutableMapping[str, typing.Any]]\n            send_stream, recv_stream = anyio.create_memory_object_stream()\n\n            async def receive_or_disconnect() -> Message:\n                if response_sent.is_set():\n                    return {\"type\": \"http.disconnect\"}\n\n                async with anyio.create_task_group() as task_group:\n\n                    async def wrap(func: typing.Callable[[], typing.Awaitable[T]]) -> T:\n                        result = await func()\n                        task_group.cancel_scope.cancel()\n                        return result\n\n                    task_group.start_soon(wrap, response_sent.wait)\n                    message = await wrap(wrapped_receive)\n\n                if response_sent.is_set():\n                    return {\"type\": \"http.disconnect\"}\n\n                return message\n\n            async def close_recv_stream_on_response_sent() -> None:\n                await response_sent.wait()\n                recv_stream.close()\n\n            async def send_no_error(message: Message) -> None:\n                try:\n                    await send_stream.send(message)\n                except anyio.BrokenResourceError:\n                    # recv_stream has been closed, i.e. response_sent has been set.\n                    return\n\n            async def coro() -> None:\n                nonlocal app_exc\n\n                async with send_stream:\n                    try:\n                        await self.app(scope, receive_or_disconnect, send_no_error)\n                    except Exception as exc:\n                        app_exc = exc\n\n            task_group.start_soon(close_recv_stream_on_response_sent)\n            task_group.start_soon(coro)\n\n            try:\n                message = await recv_stream.receive()\n                info = message.get(\"info\", None)\n                if message[\"type\"] == \"http.response.debug\" and info is not None:\n                    message = await recv_stream.receive()\n            except anyio.EndOfStream:\n                if app_exc is not None:\n                    raise app_exc\n                raise RuntimeError(\"No response returned.\")\n\n            assert message[\"type\"] == \"http.response.start\"\n\n            async def body_stream() -> typing.AsyncGenerator[bytes, None]:\n                async with recv_stream:\n                    async for message in recv_stream:\n                        assert message[\"type\"] == \"http.response.body\"\n                        body = message.get(\"body\", b\"\")\n                        if body:\n                            yield body\n                        if not message.get(\"more_body\", False):\n                            break\n\n                if app_exc is not None:\n                    raise app_exc\n\n            response = _StreamingResponse(\n                status_code=message[\"status\"], content=body_stream(), info=info\n            )\n            response.raw_headers = message[\"headers\"]\n            return response\n\n        with _convert_excgroups():\n            async with anyio.create_task_group() as task_group:\n                response = await self.dispatch_func(request, call_next)\n                await response(scope, wrapped_receive, send)\n                response_sent.set()\n\n    async def dispatch(\n        self, request: Request, call_next: RequestResponseEndpoint\n    ) -> Response:\n        raise NotImplementedError()  # pragma: no cover\n\n\nclass _StreamingResponse(StreamingResponse):\n    def __init__(\n        self,\n        content: ContentStream,\n        status_code: int = 200,\n        headers: typing.Optional[typing.Mapping[str, str]] = None,\n        media_type: typing.Optional[str] = None,\n        background: typing.Optional[BackgroundTask] = None,\n        info: typing.Optional[typing.Mapping[str, typing.Any]] = None,\n    ) -> None:\n        self._info = info\n        super().__init__(content, status_code, headers, media_type, background)\n\n    async def stream_response(self, send: Send) -> None:\n        if self._info:\n            await send({\"type\": \"http.response.debug\", \"info\": self._info})\n        return await super().stream_response(send)\n", "path": "starlette/middleware/base.py"}], "after_files": [{"content": "import asyncio\nimport functools\nimport sys\nimport typing\nfrom contextlib import contextmanager\n\nif sys.version_info >= (3, 10):  # pragma: no cover\n    from typing import TypeGuard\nelse:  # pragma: no cover\n    from typing_extensions import TypeGuard\n\nhas_exceptiongroups = True\nif sys.version_info < (3, 11):  # pragma: no cover\n    try:\n        from exceptiongroup import BaseExceptionGroup\n    except ImportError:\n        has_exceptiongroups = False\n\nT = typing.TypeVar(\"T\")\nAwaitableCallable = typing.Callable[..., typing.Awaitable[T]]\n\n\n@typing.overload\ndef is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]:\n    ...\n\n\n@typing.overload\ndef is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]:\n    ...\n\n\ndef is_async_callable(obj: typing.Any) -> typing.Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return asyncio.iscoroutinefunction(obj) or (\n        callable(obj) and asyncio.iscoroutinefunction(obj.__call__)\n    )\n\n\nT_co = typing.TypeVar(\"T_co\", covariant=True)\n\n\nclass AwaitableOrContextManager(\n    typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]\n):\n    ...\n\n\nclass SupportsAsyncClose(typing.Protocol):\n    async def close(self) -> None:\n        ...  # pragma: no cover\n\n\nSupportsAsyncCloseType = typing.TypeVar(\n    \"SupportsAsyncCloseType\", bound=SupportsAsyncClose, covariant=False\n)\n\n\nclass AwaitableOrContextManagerWrapper(typing.Generic[SupportsAsyncCloseType]):\n    __slots__ = (\"aw\", \"entered\")\n\n    def __init__(self, aw: typing.Awaitable[SupportsAsyncCloseType]) -> None:\n        self.aw = aw\n\n    def __await__(self) -> typing.Generator[typing.Any, None, SupportsAsyncCloseType]:\n        return self.aw.__await__()\n\n    async def __aenter__(self) -> SupportsAsyncCloseType:\n        self.entered = await self.aw\n        return self.entered\n\n    async def __aexit__(self, *args: typing.Any) -> typing.Union[None, bool]:\n        await self.entered.close()\n        return None\n\n\n@contextmanager\ndef collapse_excgroups() -> typing.Generator[None, None, None]:\n    try:\n        yield\n    except BaseException as exc:\n        if has_exceptiongroups:\n            while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:\n                exc = exc.exceptions[0]  # pragma: no cover\n\n        raise exc\n", "path": "starlette/_utils.py"}, {"content": "import typing\n\nimport anyio\nfrom anyio.abc import ObjectReceiveStream, ObjectSendStream\n\nfrom starlette._utils import collapse_excgroups\nfrom starlette.background import BackgroundTask\nfrom starlette.requests import ClientDisconnect, Request\nfrom starlette.responses import ContentStream, Response, StreamingResponse\nfrom starlette.types import ASGIApp, Message, Receive, Scope, Send\n\nRequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\nDispatchFunction = typing.Callable[\n    [Request, RequestResponseEndpoint], typing.Awaitable[Response]\n]\nT = typing.TypeVar(\"T\")\n\n\nclass _CachedRequest(Request):\n    \"\"\"\n    If the user calls Request.body() from their dispatch function\n    we cache the entire request body in memory and pass that to downstream middlewares,\n    but if they call Request.stream() then all we do is send an\n    empty body so that downstream things don't hang forever.\n    \"\"\"\n\n    def __init__(self, scope: Scope, receive: Receive):\n        super().__init__(scope, receive)\n        self._wrapped_rcv_disconnected = False\n        self._wrapped_rcv_consumed = False\n        self._wrapped_rc_stream = self.stream()\n\n    async def wrapped_receive(self) -> Message:\n        # wrapped_rcv state 1: disconnected\n        if self._wrapped_rcv_disconnected:\n            # we've already sent a disconnect to the downstream app\n            # we don't need to wait to get another one\n            # (although most ASGI servers will just keep sending it)\n            return {\"type\": \"http.disconnect\"}\n        # wrapped_rcv state 1: consumed but not yet disconnected\n        if self._wrapped_rcv_consumed:\n            # since the downstream app has consumed us all that is left\n            # is to send it a disconnect\n            if self._is_disconnected:\n                # the middleware has already seen the disconnect\n                # since we know the client is disconnected no need to wait\n                # for the message\n                self._wrapped_rcv_disconnected = True\n                return {\"type\": \"http.disconnect\"}\n            # we don't know yet if the client is disconnected or not\n            # so we'll wait until we get that message\n            msg = await self.receive()\n            if msg[\"type\"] != \"http.disconnect\":  # pragma: no cover\n                # at this point a disconnect is all that we should be receiving\n                # if we get something else, things went wrong somewhere\n                raise RuntimeError(f\"Unexpected message received: {msg['type']}\")\n            return msg\n\n        # wrapped_rcv state 3: not yet consumed\n        if getattr(self, \"_body\", None) is not None:\n            # body() was called, we return it even if the client disconnected\n            self._wrapped_rcv_consumed = True\n            return {\n                \"type\": \"http.request\",\n                \"body\": self._body,\n                \"more_body\": False,\n            }\n        elif self._stream_consumed:\n            # stream() was called to completion\n            # return an empty body so that downstream apps don't hang\n            # waiting for a disconnect\n            self._wrapped_rcv_consumed = True\n            return {\n                \"type\": \"http.request\",\n                \"body\": b\"\",\n                \"more_body\": False,\n            }\n        else:\n            # body() was never called and stream() wasn't consumed\n            try:\n                stream = self.stream()\n                chunk = await stream.__anext__()\n                self._wrapped_rcv_consumed = self._stream_consumed\n                return {\n                    \"type\": \"http.request\",\n                    \"body\": chunk,\n                    \"more_body\": not self._stream_consumed,\n                }\n            except ClientDisconnect:\n                self._wrapped_rcv_disconnected = True\n                return {\"type\": \"http.disconnect\"}\n\n\nclass BaseHTTPMiddleware:\n    def __init__(\n        self, app: ASGIApp, dispatch: typing.Optional[DispatchFunction] = None\n    ) -> None:\n        self.app = app\n        self.dispatch_func = self.dispatch if dispatch is None else dispatch\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        if scope[\"type\"] != \"http\":\n            await self.app(scope, receive, send)\n            return\n\n        request = _CachedRequest(scope, receive)\n        wrapped_receive = request.wrapped_receive\n        response_sent = anyio.Event()\n\n        async def call_next(request: Request) -> Response:\n            app_exc: typing.Optional[Exception] = None\n            send_stream: ObjectSendStream[typing.MutableMapping[str, typing.Any]]\n            recv_stream: ObjectReceiveStream[typing.MutableMapping[str, typing.Any]]\n            send_stream, recv_stream = anyio.create_memory_object_stream()\n\n            async def receive_or_disconnect() -> Message:\n                if response_sent.is_set():\n                    return {\"type\": \"http.disconnect\"}\n\n                async with anyio.create_task_group() as task_group:\n\n                    async def wrap(func: typing.Callable[[], typing.Awaitable[T]]) -> T:\n                        result = await func()\n                        task_group.cancel_scope.cancel()\n                        return result\n\n                    task_group.start_soon(wrap, response_sent.wait)\n                    message = await wrap(wrapped_receive)\n\n                if response_sent.is_set():\n                    return {\"type\": \"http.disconnect\"}\n\n                return message\n\n            async def close_recv_stream_on_response_sent() -> None:\n                await response_sent.wait()\n                recv_stream.close()\n\n            async def send_no_error(message: Message) -> None:\n                try:\n                    await send_stream.send(message)\n                except anyio.BrokenResourceError:\n                    # recv_stream has been closed, i.e. response_sent has been set.\n                    return\n\n            async def coro() -> None:\n                nonlocal app_exc\n\n                async with send_stream:\n                    try:\n                        await self.app(scope, receive_or_disconnect, send_no_error)\n                    except Exception as exc:\n                        app_exc = exc\n\n            task_group.start_soon(close_recv_stream_on_response_sent)\n            task_group.start_soon(coro)\n\n            try:\n                message = await recv_stream.receive()\n                info = message.get(\"info\", None)\n                if message[\"type\"] == \"http.response.debug\" and info is not None:\n                    message = await recv_stream.receive()\n            except anyio.EndOfStream:\n                if app_exc is not None:\n                    raise app_exc\n                raise RuntimeError(\"No response returned.\")\n\n            assert message[\"type\"] == \"http.response.start\"\n\n            async def body_stream() -> typing.AsyncGenerator[bytes, None]:\n                async with recv_stream:\n                    async for message in recv_stream:\n                        assert message[\"type\"] == \"http.response.body\"\n                        body = message.get(\"body\", b\"\")\n                        if body:\n                            yield body\n                        if not message.get(\"more_body\", False):\n                            break\n\n                if app_exc is not None:\n                    raise app_exc\n\n            response = _StreamingResponse(\n                status_code=message[\"status\"], content=body_stream(), info=info\n            )\n            response.raw_headers = message[\"headers\"]\n            return response\n\n        with collapse_excgroups():\n            async with anyio.create_task_group() as task_group:\n                response = await self.dispatch_func(request, call_next)\n                await response(scope, wrapped_receive, send)\n                response_sent.set()\n\n    async def dispatch(\n        self, request: Request, call_next: RequestResponseEndpoint\n    ) -> Response:\n        raise NotImplementedError()  # pragma: no cover\n\n\nclass _StreamingResponse(StreamingResponse):\n    def __init__(\n        self,\n        content: ContentStream,\n        status_code: int = 200,\n        headers: typing.Optional[typing.Mapping[str, str]] = None,\n        media_type: typing.Optional[str] = None,\n        background: typing.Optional[BackgroundTask] = None,\n        info: typing.Optional[typing.Mapping[str, typing.Any]] = None,\n    ) -> None:\n        self._info = info\n        super().__init__(content, status_code, headers, media_type, background)\n\n    async def stream_response(self, send: Send) -> None:\n        if self._info:\n            await send({\"type\": \"http.response.debug\", \"info\": self._info})\n        return await super().stream_response(send)\n", "path": "starlette/middleware/base.py"}]}