Proposal: Custom Request Classes using type annotations

starlette/endpoints.py

@@ -22,9 +22,17 @@ def __await__(self) -> typing.Generator:
         return self.dispatch().__await__()
 
     async def dispatch(self) -> None:
-        request = Request(self.scope, receive=self.receive)
-        handler_name = "get" if request.method == "HEAD" else request.method.lower()
+        method = self.scope["method"]
+        handler_name = "get" if method == "HEAD" else method.lower()
         handler = getattr(self, handler_name, self.method_not_allowed)
+
+        request_class = handler.__annotations__.get("request", Request)
+        if not issubclass(request_class, Request):
+            raise TypeError(
+                "Custom Request classes must subclass `starlette.requests.Request`"
+            )
+
+        request = request_class(self.scope, receive=self.receive)
         is_async = asyncio.iscoroutinefunction(handler)
         if is_async:
             response = await handler(request)

starlette/exceptions.py

@@ -85,7 +85,13 @@ async def sender(message: Message) -> None:
                 msg = "Caught handled exception, but response already started."
                 raise RuntimeError(msg) from exc
 
-            request = Request(scope, receive=receive)
+            request_class = handler.__annotations__.get("request", Request)
+            if not issubclass(request_class, Request):
+                raise TypeError(
+                    "Custom Request classes must subclass `starlette.requests.Request`"
+                )
+
+            request = request_class(scope, receive=receive)
             if asyncio.iscoroutinefunction(handler):
                 response = await handler(request, exc)
             else:

starlette/middleware/base.py

@@ -21,7 +21,13 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             await self.app(scope, receive, send)
             return
 
-        request = Request(scope, receive=receive)
+        request_class = self.dispatch_func.__annotations__.get("request", Request)
+        if not issubclass(request_class, Request):
+            raise TypeError(
+                "Custom Request classes must subclass `starlette.requests.Request`"
+            )
+
+        request = request_class(scope, receive=receive)
         response = await self.dispatch_func(request, self.call_next)
         await response(scope, receive, send)
 

starlette/routing.py

@@ -36,7 +36,13 @@ def request_response(func: typing.Callable) -> ASGIApp:
     is_coroutine = asyncio.iscoroutinefunction(func)
 
     async def app(scope: Scope, receive: Receive, send: Send) -> None:
-        request = Request(scope, receive=receive, send=send)
+        request_class = func.__annotations__.get("request", Request)
+        if not issubclass(request_class, Request):
+            raise TypeError(
+                "Custom Request classes must subclass `starlette.requests.Request`"
+            )
+
+        request = request_class(scope, receive=receive, send=send)
         if is_coroutine:
             response = await func(request)
         else:

tests/middleware/test_base.py

@@ -3,6 +3,7 @@
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
 from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
 from starlette.responses import PlainTextResponse
 from starlette.routing import Route
 from starlette.testclient import TestClient
@@ -29,6 +30,14 @@ def exc(request):
     raise Exception()
 
 
+class CustomRequest(Request):
+    pass
+
+
+class BadCustomRequest:
+    pass
+
+
 @app.route("/no-response")
 class NoResponse:
     def __init__(self, scope, receive, send):
@@ -77,6 +86,7 @@ async def plaintext(request, call_next):
             return PlainTextResponse("OK")
         response = await call_next(request)
         response.headers["Custom"] = "Example"
+        response.headers["X-Request-Class"] = request.__class__.__name__
         return response
 
     client = TestClient(app)
@@ -86,6 +96,47 @@ async def plaintext(request, call_next):
     response = client.get("/homepage")
     assert response.text == "Homepage"
     assert response.headers["Custom"] == "Example"
+    assert response.headers["X-Request-Class"] == "Request"
+
+
+def test_middleware_decorator_custom_req():
+    app = Starlette()
+
+    @app.route("/custom-request")
+    def custom_req(request):
+        return PlainTextResponse(request.__class__.__name__)
+
+    @app.middleware("http")
+    async def plaintext(request: CustomRequest, call_next):
+        response = await call_next(request)
+        response.headers["Custom"] = "Example"
+        response.headers["X-Request-Class"] = request.__class__.__name__
+        return response
+
+    client = TestClient(app)
+
+    response = client.get("/custom-request")
+    assert response.text == "Request"
+    assert response.headers["X-Request-Class"] == "CustomRequest"
+
+
+def test_middleware_decorator_custom_req_fails():
+    app = Starlette()
+
+    @app.route("/bad-custom-request")
+    def custom_req(request):  # pragma: no cover
+        return PlainTextResponse(request.__class__.__name__)
+
+    @app.middleware("http")
+    async def plaintext(request: BadCustomRequest, call_next):  # pragma: no cover
+        response = await call_next(request)
+        response.headers["Custom"] = "Example"
+        response.headers["X-Request-Class"] = request.__class__.__name__
+        return response
+
+    client = TestClient(app)
+    with pytest.raises(TypeError):
+        client.get("/bad-custom-request")
 
 
 def test_state_data_across_multiple_middlewares():

tests/test_endpoints.py

@@ -1,6 +1,7 @@
 import pytest
 
 from starlette.endpoints import HTTPEndpoint, WebSocketEndpoint
+from starlette.requests import Request
 from starlette.responses import PlainTextResponse
 from starlette.routing import Route, Router
 from starlette.testclient import TestClient
@@ -14,8 +15,28 @@ async def get(self, request):
         return PlainTextResponse(f"Hello, {username}!")
 
 
+class CustomRequest(Request):
+    pass
+
+
+class BadCustomRequest:
+    pass
+
+
+class CustomRequestClasses(HTTPEndpoint):
+    async def get(self, request: CustomRequest):
+        return PlainTextResponse(request.__class__.__name__)
+
+    async def post(self, request: BadCustomRequest):  # pragma: no cover
+        return PlainTextResponse(request.__class__.__name__)
+
+
 app = Router(
-    routes=[Route("/", endpoint=Homepage), Route("/{username}", endpoint=Homepage)]
+    routes=[
+        Route("/", endpoint=Homepage),
+        Route("/custom-request", endpoint=CustomRequestClasses),
+        Route("/{username}", endpoint=Homepage),
+    ]
 )
 
 client = TestClient(app)
@@ -39,6 +60,17 @@ def test_http_endpoint_route_method():
     assert response.text == "Method Not Allowed"
 
 
+def test_http_endpoint_custom_req():
+    response = client.get("/custom-request")
+    assert response.status_code == 200
+    assert response.text == "CustomRequest"
+
+
+def test_http_endpoint_custom_req_fail():
+    with pytest.raises(TypeError):
+        client.post("/custom-request")
+
+
 def test_websocket_endpoint_on_connect():
     class WebSocketApp(WebSocketEndpoint):
         async def on_connect(self, websocket):

tests/test_exceptions.py

@@ -1,6 +1,7 @@
 import pytest
 
 from starlette.exceptions import ExceptionMiddleware, HTTPException
+from starlette.requests import Request
 from starlette.responses import PlainTextResponse
 from starlette.routing import Route, Router, WebSocketRoute
 from starlette.testclient import TestClient
@@ -18,6 +19,28 @@ def not_modified(request):
     raise HTTPException(status_code=304)
 
 
+def sad_server(request):
+    if request.headers.get("except") == "503":
+        raise HTTPException(status_code=503)
+    raise HTTPException(status_code=504)
+
+
+class CustomRequest(Request):
+    pass
+
+
+class BadCustomRequest:
+    pass
+
+
+def fine_exception_handler(request: CustomRequest, exc):
+    return PlainTextResponse(request.__class__.__name__, status_code=exc.status_code)
+
+
+def bad_exception_handler(request: BadCustomRequest, exc):  # pragma: no cover
+    return PlainTextResponse(request.__class__.__name__, status_code=exc.status_code)
+
+
 class HandledExcAfterResponse:
     async def __call__(self, scope, receive, send):
         response = PlainTextResponse("OK", status_code=200)
@@ -31,12 +54,15 @@ async def __call__(self, scope, receive, send):
         Route("/not_acceptable", endpoint=not_acceptable),
         Route("/not_modified", endpoint=not_modified),
         Route("/handled_exc_after_response", endpoint=HandledExcAfterResponse()),
+        Route("/sad_server", endpoint=sad_server),
         WebSocketRoute("/runtime_error", endpoint=raise_runtime_error),
     ]
 )
 
 
-app = ExceptionMiddleware(router)
+app = ExceptionMiddleware(
+    router, handlers={503: fine_exception_handler, 504: bad_exception_handler}
+)
 client = TestClient(app)
 
 
@@ -95,3 +121,14 @@ class CustomHTTPException(HTTPException):
     assert repr(CustomHTTPException(500, detail="Something custom")) == (
         "CustomHTTPException(status_code=500, detail='Something custom')"
     )
+
+
+def test_sad_server_custom_req():
+    response = client.get("/sad_server", headers={"except": "503"})
+    assert response.status_code == 503
+    assert response.text == "CustomRequest"
+
+
+def test_sad_server_custom_req_fail():
+    with pytest.raises(TypeError):
+        client.get("/sad_server")

tests/test_routing.py

@@ -3,6 +3,7 @@
 import pytest
 
 from starlette.applications import Starlette
+from starlette.requests import Request
 from starlette.responses import JSONResponse, PlainTextResponse, Response
 from starlette.routing import Host, Mount, NoMatchFound, Route, Router, WebSocketRoute
 from starlette.testclient import TestClient
@@ -32,6 +33,22 @@ def user_no_match(request):  # pragma: no cover
     return Response(content, media_type="text/plain")
 
 
+class CustomRequest(Request):
+    pass
+
+
+class BadCustomRequest:
+    pass
+
+
+def custom_request(request: CustomRequest):
+    return Response(request.__class__.__name__, media_type="text/plain")
+
+
+def custom_request_exc(request: BadCustomRequest):  # pragma: no cover
+    return Response(request.__class__.__name__, media_type="text/plain")
+
+
 app = Router(
     [
         Route("/", endpoint=homepage, methods=["GET"]),
@@ -44,6 +61,8 @@ def user_no_match(request):  # pragma: no cover
                 Route("/nomatch", endpoint=user_no_match),
             ],
         ),
+        Route("/custom-request", endpoint=custom_request),
+        Route("/bad-custom-request", endpoint=custom_request_exc),
         Mount("/static", app=Response("xxxxx", media_type="image/png")),
     ]
 )
@@ -134,6 +153,13 @@ def test_router():
     assert response.status_code == 200
     assert response.text == "User nomatch"
 
+    response = client.get("/custom-request")
+    assert response.status_code == 200
+    assert response.text == "CustomRequest"
+
+    with pytest.raises(TypeError):
+        response = client.get("/bad-custom-request")
+
     response = client.get("/static/123")
     assert response.status_code == 200
     assert response.text == "xxxxx"