fix(params): support typed collection casting (#409)

Author: tarsil

docs/en/docs/release-notes.md

@@ -1,5 +1,14 @@
 # Release Notes
 
+## 0.26.2
+
+### Fixed
+
+- Query, header, and cookie parameter casting now handles repeated values correctly, preserving list-style inputs instead of nesting them.
+- Parameter casting now supports typed collection casts such as `list[str]`, `dict[str, int]`, and nested union collections like `list[str | int | dict[str, str]]`.
+- `cast=list` now treats a single string value as a one-item list instead of splitting it into characters.
+- Default parameter values are now passed through their configured cast when applicable.
+
 ## 0.26.1
 
 ### Added

lilya/__init__.py

@@ -1 +1 @@
-__version__ = "0.26.1"
+__version__ = "0.26.2"

lilya/_internal/_responses.py

@@ -37,7 +37,7 @@
     UnprocessableEntity,
     WebSocketException,
 )
-from lilya.params import Cookie, Header, Query
+from lilya.params import Cookie, Header, Query, get_cast_name
 from lilya.requests import Request
 from lilya.responses import Ok, Response
 from lilya.serializers import serializer
@@ -59,6 +59,22 @@
 _SIG_CACHE_ATTR = "__lilya_resolved_signature__"
 _PLAN_CACHE_ATTR = "__lilya_handler_plan__"
 
+
+def _resolve_bound_param_value(field: Query | Header | Cookie, raw_value: Any) -> Any:
+    if raw_value is None:
+        default = getattr(field, "default", None)
+        return (
+            field.resolve(default, field.cast) if field.cast and default is not None else default
+        )
+
+    if field.cast:
+        return field.resolve(raw_value, field.cast)
+
+    if isinstance(raw_value, list) and len(raw_value) == 1:
+        return raw_value[0]
+    return raw_value
+
+
 ZeroArgAsyncHandler = Callable[[], Awaitable[Any]]
 KwargsAsyncHandler = Callable[..., Awaitable[Any]]
 ZeroArgSyncHandler = Callable[[], Any]
@@ -486,24 +502,13 @@ def get_request() -> Request:
                         if field.required and raw_value is None:
                             raise UnprocessableEntity(f"Missing mandatory query parameter '{key}'")
 
-                        if raw_value is None:
-                            fast_params[n] = getattr(field, "default", None)
-                        else:
-                            try:
-                                if field.cast and isinstance(raw_value, list):
-                                    fast_params[n] = [raw_value]
-                                elif field.cast:
-                                    fast_params[n] = field.resolve(raw_value, field.cast)
-                                else:
-                                    fast_params[n] = (
-                                        raw_value[0]
-                                        if isinstance(raw_value, list) and len(raw_value) == 1
-                                        else raw_value
-                                    )
-                            except (TypeError, ValueError):
-                                raise UnprocessableEntity(
-                                    f"Invalid value for query parameter '{key}': expected {field.cast.__name__}"
-                                ) from None
+                        try:
+                            fast_params[n] = _resolve_bound_param_value(field, raw_value)
+                        except (TypeError, ValueError):
+                            raise UnprocessableEntity(
+                                f"Invalid value for query parameter '{key}': "
+                                f"expected {get_cast_name(field.cast)}"
+                            ) from None
 
                     # 3) Context (if requested)
                     if needs_context:
@@ -652,25 +657,12 @@ def extract_request_params_information(
             if field.required and raw_value is None:
                 raise UnprocessableEntity(f"Missing mandatory query parameter '{key}'") from None
 
-            # Fallback to default
-            if raw_value is None:
-                request_params[name] = field.default if hasattr(field, "default") else None
-                continue
-
-            # Apply casting if defined
             try:
-                if field.cast and isinstance(raw_value, list):
-                    request_params[name] = [raw_value]
-                elif field.cast:
-                    request_params[name] = field.resolve(raw_value, field.cast)
-                else:
-                    if isinstance(raw_value, list) and len(raw_value) == 1:
-                        request_params[name] = raw_value[0]
-                    else:
-                        request_params[name] = raw_value
+                request_params[name] = _resolve_bound_param_value(field, raw_value)
             except (TypeError, ValueError):
                 raise UnprocessableEntity(
-                    f"Invalid value for query parameter '{key}': expected {field.cast.__name__}"
+                    f"Invalid value for query parameter '{key}': "
+                    f"expected {get_cast_name(field.cast)}"
                 ) from None
 
         return request_params

lilya/params.py

@@ -1,31 +1,134 @@
+import json
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any
+from types import UnionType
+from typing import Annotated, Any, Union, get_args, get_origin
+
+TRUE_VALUES = {"true", "1", "yes", "on", "t"}
+FALSE_VALUES = {"false", "0", "no", "off", "f"}
+
+
+def _strip_annotated(cast: Any) -> Any:
+    while get_origin(cast) is Annotated:
+        cast = get_args(cast)[0]
+    return cast
+
+
+def _is_union(cast: Any) -> bool:
+    return get_origin(cast) in (Union, UnionType)
+
+
+def _as_list(value: Any) -> list[Any]:
+    if isinstance(value, list):
+        return value
+    if isinstance(value, tuple | set | frozenset):
+        return list(value)
+    return [value]
+
+
+def _as_mapping(value: Any) -> Mapping[Any, Any]:
+    if isinstance(value, Mapping):
+        return value
+
+    if isinstance(value, bytes):
+        value = value.decode("utf-8")
+
+    if isinstance(value, str):
+        value = json.loads(value)
+        if isinstance(value, Mapping):
+            return value
+
+    return dict(value)
+
+
+def get_cast_name(cast: Any) -> str:
+    return getattr(cast, "__name__", str(cast).replace("typing.", ""))
 
 
 @dataclass
 class BaseParam:
-    def __cast__(self, value: Any, cast: type) -> Any:
-        try:
-            if str(value).lower() in ("true", "1", "yes", "on", "t") and cast is bool:
-                value = True
-            elif str(value).lower() in ("false", "0", "no", "off", "f") and cast is bool:
-                value = False
-            else:
-                value = cast(value)
-        except Exception:
-            raise
-        return value
+    def __cast__(self, value: Any, cast: Any) -> Any:
+        cast = _strip_annotated(cast)
+
+        if cast is Any or cast is object:
+            return value
+
+        if _is_union(cast):
+            union_types = [typ for typ in get_args(cast) if typ is not type(None)]
+            if value is None and len(union_types) != len(get_args(cast)):
+                return None
+
+            if isinstance(value, str) and str in union_types and len(union_types) > 1:
+                union_types = [typ for typ in union_types if typ is not str] + [str]
+
+            for typ in union_types:
+                try:
+                    return self.__cast__(value, typ)
+                except (TypeError, ValueError):
+                    continue
+            raise ValueError(f"Cannot cast value {value!r} to {get_cast_name(cast)}")
+
+        origin = get_origin(cast)
+        args = get_args(cast)
+
+        if cast is bool:
+            if isinstance(value, bool):
+                return value
+            if isinstance(value, str):
+                value_lower = value.lower()
+                if value_lower in TRUE_VALUES:
+                    return True
+                if value_lower in FALSE_VALUES:
+                    return False
+                raise ValueError(f"Cannot cast value {value!r} to bool")
+            return bool(value)
+
+        if cast is list or origin is list:
+            values = _as_list(value)
+            item_cast = args[0] if args else Any
+            return [self.__cast__(item, item_cast) for item in values]
+
+        if cast is tuple or origin is tuple:
+            values = _as_list(value)
+            if not args:
+                return tuple(values)
+            if len(args) == 2 and args[1] is Ellipsis:
+                return tuple(self.__cast__(item, args[0]) for item in values)
+            return tuple(
+                self.__cast__(item, item_cast)
+                for item, item_cast in zip(values, args, strict=False)
+            )
+
+        if cast is set or origin is set:
+            values = _as_list(value)
+            item_cast = args[0] if args else Any
+            return {self.__cast__(item, item_cast) for item in values}
+
+        if cast is frozenset or origin is frozenset:
+            values = _as_list(value)
+            item_cast = args[0] if args else Any
+            return frozenset(self.__cast__(item, item_cast) for item in values)
+
+        if cast is dict or origin is dict:
+            mapping = _as_mapping(value)
+            key_cast, value_cast = args or (Any, Any)
+            return {
+                self.__cast__(key, key_cast): self.__cast__(item, value_cast)
+                for key, item in mapping.items()
+            }
+
+        return cast(value)
 
 
 @dataclass
 class Query(BaseParam):
     default: Any | None = None
     alias: str | None = None
     required: bool = False
-    cast: type | None = None
+    cast: Any | None = None
     description: str | None = None
 
-    def resolve(self, value: Any, cast: type) -> Any:
+    def resolve(self, value: Any, cast: Any) -> Any:
         return self.__cast__(value, cast) if cast else value
 
 
@@ -34,10 +137,10 @@ class Header(BaseParam):
     value: Any
     alias: str | None = None
     required: bool = False
-    cast: type | None = None
+    cast: Any | None = None
     description: str | None = None
 
-    def resolve(self, value: Any, cast: type) -> Any:
+    def resolve(self, value: Any, cast: Any) -> Any:
         return self.__cast__(value, cast) if cast else value
 
 
@@ -46,8 +149,8 @@ class Cookie(BaseParam):
     value: Any
     alias: str | None = None
     required: bool = False
-    cast: type | None = None
+    cast: Any | None = None
     description: str | None = None
 
-    def resolve(self, value: Any, cast: type) -> Any:
+    def resolve(self, value: Any, cast: Any) -> Any:
         return self.__cast__(value, cast) if cast else value

tests/params/test_cookie.py

@@ -29,6 +29,16 @@ async def cookie_casted(visitor: int = Cookie(value="visit_count", cast=int)):
     return {"visitor": visitor}
 
 
+async def cookie_typed_list(scope: list[str] = Cookie(value="scope", cast=list[str])):
+    return {"scope": scope}
+
+
+async def cookie_typed_dict(
+    settings: dict[str, bool] = Cookie(value="settings", cast=dict[str, bool]),
+):
+    return {"settings": settings}
+
+
 async def cookie_invalid_cast(visitor: int = Cookie(value="visit_count", cast=int)):
     return {"visitor": visitor}
 
@@ -82,6 +92,22 @@ def test_cookie_casted(test_client_factory):
         assert response.json() == {"visitor": 5}
 
 
+def test_cookie_typed_list(test_client_factory):
+    with create_client(
+        routes=[Path("/", cookie_typed_list)], settings_module=EncoderSettings
+    ) as client:
+        response = client.get("/", cookies={"scope": "read"})
+        assert response.json() == {"scope": ["read"]}
+
+
+def test_cookie_typed_dict(test_client_factory):
+    with create_client(
+        routes=[Path("/", cookie_typed_dict)], settings_module=EncoderSettings
+    ) as client:
+        response = client.get("/", cookies={"settings": '{"dark":"true","beta":false}'})
+        assert response.json() == {"settings": {"dark": True, "beta": False}}
+
+
 def test_cookie_invalid_cast(test_client_factory):
     with create_client(
         routes=[Path("/", cookie_invalid_cast)], settings_module=EncoderSettings

tests/params/test_headers.py

@@ -29,6 +29,14 @@ async def header_casted(content_length: int = Header(value="Content-Length", cas
     return {"length": content_length}
 
 
+async def header_typed_list(role: list[str] = Header(value="X-ROLE", cast=list[str])):
+    return {"role": role}
+
+
+async def header_typed_dict(meta: dict[str, int] = Header(value="X-META", cast=dict[str, int])):
+    return {"meta": meta}
+
+
 async def header_invalid_cast(content_length: int = Header(value="Content-Length", cast=int)):
     return {"length": content_length}
 
@@ -66,6 +74,33 @@ def test_header_casted():
         assert response.json() == {"length": 123}
 
 
+def test_header_typed_list_single(test_client_factory):
+    with create_client(
+        routes=[Path("/", header_typed_list)], settings_module=EncoderSettings
+    ) as client:
+        response = client.get("/", headers={"X-ROLE": "admin"})
+
+        assert response.json() == {"role": ["admin"]}
+
+
+def test_header_typed_list_multiple(test_client_factory):
+    with create_client(
+        routes=[Path("/", header_typed_list)], settings_module=EncoderSettings
+    ) as client:
+        response = client.get("/", headers=[("X-ROLE", "admin"), ("X-ROLE", "staff")])
+
+        assert response.json() == {"role": ["admin", "staff"]}
+
+
+def test_header_typed_dict(test_client_factory):
+    with create_client(
+        routes=[Path("/", header_typed_dict)], settings_module=EncoderSettings
+    ) as client:
+        response = client.get("/", headers={"X-META": '{"page":"2","limit":10}'})
+
+        assert response.json() == {"meta": {"page": 2, "limit": 10}}
+
+
 def test_header_invalid_cast(test_client_factory):
     with create_client(
         routes=[Path("/", header_invalid_cast)], settings_module=EncoderSettings