(
self, send: ASGISendCallable, code: int, reason: str
)
| 336 | await send(message) |
| 337 | |
| 338 | async def close_connection( |
| 339 | self, send: ASGISendCallable, code: int, reason: str |
| 340 | ) -> None: |
| 341 | if self._closed: |
| 342 | raise RuntimeError("Cannot close websocket multiple times") |
| 343 | |
| 344 | spec_version = _convert_version( |
| 345 | self.scope.get("asgi", {}).get("spec_version", "2.0") |
| 346 | ) |
| 347 | if spec_version >= [2, 3]: |
| 348 | await send({"type": "websocket.close", "code": code, "reason": reason}) |
| 349 | else: |
| 350 | await send({"type": "websocket.close", "code": code}) # type: ignore |
| 351 | self._closed = True |
| 352 | |
| 353 | |
| 354 | class ASGILifespan: |
nothing calls this directly
no test coverage detected