Takes a function or coroutine `func(request) -> response`, and returns an ASGI application.
(
func: Callable[[Request], Awaitable[Response] | Response],
)
| 44 | |
| 45 | |
| 46 | def request_response( |
| 47 | func: Callable[[Request], Awaitable[Response] | Response], |
| 48 | ) -> ASGIApp: |
| 49 | """ |
| 50 | Takes a function or coroutine `func(request) -> response`, |
| 51 | and returns an ASGI application. |
| 52 | """ |
| 53 | f: Callable[[Request], Awaitable[Response]] = ( |
| 54 | func if is_async_callable(func) else functools.partial(run_in_threadpool, func) # type: ignore[assignment, call-arg] |
| 55 | ) |
| 56 | |
| 57 | async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| 58 | request = Request(scope, receive, send) |
| 59 | |
| 60 | async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| 61 | response = await f(request) |
| 62 | await response(scope, receive, send) |
| 63 | |
| 64 | await wrap_app_handling_exceptions(app, request)(scope, receive, send) |
| 65 | |
| 66 | return app |
| 67 | |
| 68 | |
| 69 | def websocket_session( |
no test coverage detected