fix pydantic type hint check

rolo/dispatcher.py

@@ -1,8 +1,9 @@
 """
 DEPRECATED: use ``from rolo.routing import handler_dispatcher`` instead
 """
-from .routing.handler import handler_dispatcher
+from .routing.handler import ResultValue, handler_dispatcher
 
 __all__ = [
     "handler_dispatcher",
+    "ResultValue",
 ]

rolo/routing/pydantic.py

@@ -26,6 +26,8 @@ def _get_model_argument(endpoint: Handler) -> t.Optional[tuple[str, t.Type[pydan
     for arg_name, arg_type in endpoint.__annotations__.items():
         if arg_name in ("self", "return"):
             continue
+        if not inspect.isclass(arg_type):
+            continue
         if issubclass(arg_type, pydantic.BaseModel):
             return arg_name, arg_type
 

tests/serving/test_twisted.py

@@ -0,0 +1,25 @@
+import io
+
+import requests
+
+from rolo import Request, Router, route
+from rolo.dispatcher import handler_dispatcher
+from rolo.gateway import Gateway
+from rolo.gateway.handlers import RouterHandler
+
+
+def test_large_file_upload(serve_twisted_gateway):
+    router = Router(handler_dispatcher())
+
+    @route("/hello", methods=["POST"])
+    def hello(request: Request):
+        return "ok"
+
+    router.add(hello)
+
+    gateway = Gateway(request_handlers=[RouterHandler(router, True)])
+    server = serve_twisted_gateway(gateway)
+
+    response = requests.post(server.url + "/hello", io.BytesIO(b"0" * 100001))
+
+    assert response.status_code == 200

tests/test_pydantic.py

@@ -1,3 +1,5 @@
+from typing import TypedDict
+
 import pydantic
 import pytest
 from werkzeug.exceptions import BadRequest
@@ -194,3 +196,38 @@ def on_post(self, request: Request, item_id: int, item: MyItem):
             "item": {"is_offer": None, "name": "rolo", "price": 420.69},
             "item_id": 123,
         }
+
+    def test_with_generic_type_alias(self):
+        router = Router(dispatcher=handler_dispatcher())
+
+        def handler(request: Request, matrix: dict[str, str] = None):
+            return "ok"
+
+        router.add("/", endpoint=handler)
+
+        request = Request("GET", "/")
+        assert router.dispatch(request).data == b"ok"
+
+    def test_with_typed_dict(self):
+        try:
+            from typing import Unpack
+        except ImportError:
+            pytest.skip("This test only works with Python >=3.11")
+
+        router = Router(dispatcher=handler_dispatcher())
+
+        class Test(TypedDict, total=False):
+            path: str
+            random_value: str
+
+        def func(request: Request, **kwargs: Unpack[Test]):
+            return f"path={kwargs.get('path')},random_value={kwargs.get('random_value')}"
+
+        router.add(
+            "/",
+            endpoint=func,
+            defaults={"path": "", "random_value": "dev"},
+        )
+
+        request = Request("GET", "/")
+        assert router.dispatch(request).data == b"path=,random_value=dev"