Improve headers ergonomics

CHANGELOG.md

@@ -5,6 +5,31 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.6.5] - Unreleased
+
+- Improve header ergonomics: dictionary-style assignment and `Headers.update()`
+  accept strings or bytes, so `response.headers["X-Example"] = "value"` is
+  supported while headers remain stored internally as bytes. String names use
+  ASCII and string values use Latin-1; unrepresentable characters raise
+  `UnicodeEncodeError`. Assignment still replaces all case-insensitive matches.
+- Add `Headers.get_text_values(name, encoding="latin-1")` to decode matching
+  values on demand as a tuple, preserving repeated header entries.
+  `get_first_text()` returns the first value as a string, or `None` when absent;
+  `get_single_text()` returns a string and requires exactly one matching entry,
+  raising `ValueError` otherwise. Both also accept an explicit encoding.
+  Existing byte-valued reads and low-level byte-based methods remain unchanged.
+- Complete header dictionary symmetry: lookup, membership, and deletion accept
+  string or byte names; lookup always returns byte values. Add `Headers.add_text()`
+  to append without replacing matches, and `Headers.from_values()` to build a
+  byte-backed collection from mappings or iterables while preserving duplicates.
+- Use the same ASCII-name/Latin-1-value policy for HTTP client `headers=` and
+  `default_headers=`, route header filters, and testing-client iterable headers.
+  Previously, HTTP client string values required ASCII. Testing-client dictionaries
+  retain their historical UTF-8 encoding for backward compatibility.
+  Pass explicitly encoded bytes for field-specific encodings.
+- Preserve raw header values when sending HTTP/2 requests, without decoding them
+  as UTF-8, so Latin-1 values and explicitly encoded bytes reach the wire unchanged.
+
 ## [2.6.4] - 2026-10-11 :libra:
 
 - Add a `vary` parameter to the `cache_control` decorator and to

blacksheep/client/connection.py

@@ -362,44 +362,46 @@ async def connect(self) -> None:
 
             self._connected = True
 
-    def _convert_request_to_h2_headers(self, request: Request) -> list[tuple[str, str]]:
+    def _convert_request_to_h2_headers(
+        self, request: Request
+    ) -> list[tuple[bytes, bytes]]:
         """Convert a BlackSheep Request to HTTP/2 pseudo-headers and headers."""
         # HTTP/2 pseudo-headers
         path = request.url.path or b"/"
         if request.url.query:
             path = path + b"?" + request.url.query
 
         headers = [
-            (":method", request.method),
-            (":path", path.decode("utf-8")),
-            (":scheme", self._cached_scheme),
-            (":authority", self.host),
+            (b":method", request.method.encode("ascii")),
+            (b":path", path),
+            (b":scheme", self._cached_scheme.encode("ascii")),
+            (b":authority", self.host.encode("utf-8")),
         ]
 
         # Add Content-Type header from content if present
         if request.content and request.content.type:
             content_type = request.content.type
-            content_type_str = (
-                content_type.decode("utf-8")
+            content_type_bytes = (
+                content_type
                 if isinstance(content_type, bytes)
-                else content_type
+                else content_type.encode("ascii")
             )
-            headers.append(("content-type", content_type_str))
+            headers.append((b"content-type", content_type_bytes))
 
         # Add regular headers
         for name, value in request.headers:
-            name_str = (
-                name.decode("utf-8").lower()
+            name_bytes = (
+                name.lower()
                 if isinstance(name, bytes)
-                else name.lower()
+                else name.encode("ascii").lower()
             )
-            value_str = value.decode("utf-8") if isinstance(value, bytes) else value
+            value_bytes = value if isinstance(value, bytes) else value.encode("latin-1")
 
             # Skip headers that are represented as pseudo-headers in HTTP/2
-            if name_str in ("host", "connection", "transfer-encoding"):
+            if name_bytes in (b"host", b"connection", b"transfer-encoding"):
                 continue
 
-            headers.append((name_str, value_str))
+            headers.append((name_bytes, value_bytes))
 
         return headers
 
@@ -457,7 +459,7 @@ async def send(self, request: Request) -> Response:
 
             # Check for Expect: 100-continue header
             expect_continue = any(
-                (h[0] == "expect" and h[1] == "100-continue") for h in h2_headers
+                (h[0] == b"expect" and h[1] == b"100-continue") for h in h2_headers
             )
 
             # Send headers (without end_stream if expecting 100-continue with body)

blacksheep/common/types.py

@@ -2,20 +2,21 @@
 Common types annotations and functions.
 """
 
-from typing import AnyStr, Iterable
+from collections.abc import Mapping
+from typing import AnyStr, Iterable, TypeVar
 
+from blacksheep.headers import Headers
 from blacksheep.url import URL
 
 KeyValuePair = tuple[AnyStr, AnyStr]
-HeadersType = dict[AnyStr, AnyStr] | Iterable[KeyValuePair]
+_HeaderName = TypeVar("_HeaderName", bound=str | bytes)
+HeadersType = (
+    Mapping[_HeaderName, str | bytes] | Iterable[tuple[str | bytes, str | bytes]]
+)
 ParamsType = dict[AnyStr, AnyStr] | Iterable[KeyValuePair]
 URLType = str | bytes | URL
 
 
-def _ensure_header_bytes(value: AnyStr) -> bytes:
-    return value if isinstance(value, bytes) else value.encode("ascii")
-
-
 def _ensure_param_str(value: AnyStr) -> str:
     return value if isinstance(value, str) else value.decode("ascii")
 
@@ -25,15 +26,7 @@ def normalize_headers(
 ) -> list[tuple[bytes, bytes]] | None:
     if headers is None:
         return None
-    if isinstance(headers, dict):
-        return [
-            (_ensure_header_bytes(key), _ensure_header_bytes(value))  # type: ignore
-            for key, value in headers.items()
-        ]
-    return [
-        (_ensure_header_bytes(key), _ensure_header_bytes(value))
-        for key, value in headers
-    ]
+    return Headers.from_values(headers).values
 
 
 def normalize_params(params: ParamsType | None) -> list[tuple[str, str]] | None:

blacksheep/headers.py

@@ -1,5 +1,23 @@
-from collections.abc import Mapping, MutableSequence
-from typing import Generator
+from collections.abc import Iterable, Mapping, MutableSequence
+from typing import Generator, TypeVar
+
+_HeaderName = TypeVar("_HeaderName", bound=str | bytes)
+
+
+def _encode_header_name(name: str | bytes) -> bytes:
+    if isinstance(name, str):
+        return name.encode("ascii")
+    if not isinstance(name, bytes):
+        raise TypeError("Header names must be str or bytes")
+    return name
+
+
+def _encode_header_value(value: str | bytes) -> bytes:
+    if isinstance(value, str):
+        return value.encode("latin-1")
+    if not isinstance(value, bytes):
+        raise TypeError("Header values must be str or bytes")
+    return value
 
 
 class Header:
@@ -26,6 +44,27 @@ def __init__(self, values: list[tuple[bytes, bytes]] | None = None):
             values = []
         self.values = values
 
+    @classmethod
+    def from_values(
+        cls,
+        values: (
+            Mapping[_HeaderName, str | bytes]
+            | Iterable[tuple[str | bytes, str | bytes]]
+        ),
+    ) -> "Headers":
+        """Create byte-backed headers from a mapping or iterable, preserving duplicates."""
+        items = values.items() if isinstance(values, Mapping) else values
+        return cls(
+            [
+                (_encode_header_name(name), _encode_header_value(value))
+                for name, value in items
+            ]
+        )
+
+    def add_text(self, name: str | bytes, value: str | bytes) -> None:
+        """Append an entry without replacing matches, encoding strings at the boundary."""
+        self.add(_encode_header_name(name), _encode_header_value(value))
+
     def get(self, name: bytes) -> tuple[bytes, ...]:
         results = []
         name = name.lower()
@@ -34,6 +73,25 @@ def get(self, name: bytes) -> tuple[bytes, ...]:
                 results.append(header[1])
         return tuple(results)
 
+    def get_text_values(
+        self, name: str | bytes, encoding: str = "latin-1"
+    ) -> tuple[str, ...]:
+        """Decode matching values on demand, preserving separate header entries."""
+        return tuple(
+            value.decode(encoding) for value in self.get(_encode_header_name(name))
+        )
+
+    def get_first_text(
+        self, name: str | bytes, encoding: str = "latin-1"
+    ) -> str | None:
+        """Decode the first matching value, or return None when absent."""
+        value = self.get_first(_encode_header_name(name))
+        return value.decode(encoding) if value is not None else None
+
+    def get_single_text(self, name: str | bytes, encoding: str = "latin-1") -> str:
+        """Decode exactly one matching value; raise ValueError otherwise."""
+        return self.get_single(_encode_header_name(name)).decode(encoding)
+
     def get_tuples(self, name: bytes) -> list[tuple[bytes, bytes]]:
         results = []
         name = name.lower()
@@ -63,7 +121,7 @@ def merge(self, values: list[tuple[bytes, bytes]]):
                 continue
             self.values.append(header)
 
-    def update(self, values: dict[bytes, bytes]):
+    def update(self, values: Mapping[_HeaderName, bytes | str]):
         for key, value in values.items():
             self[key] = value
 
@@ -139,11 +197,12 @@ def __iadd__(
     def __iter__(self) -> Generator[tuple[bytes, bytes], None, None]:
         yield from self.values
 
-    def __setitem__(self, key: bytes, value: bytes):
-        self.set(key, value)
+    def __setitem__(self, key: str | bytes, value: str | bytes):
+        """Replace matching entries, encoding text names as ASCII and values as Latin-1."""
+        self.set(_encode_header_name(key), _encode_header_value(value))
 
-    def __getitem__(self, item: bytes) -> tuple[bytes, ...]:
-        return self.get(item)
+    def __getitem__(self, item: str | bytes) -> tuple[bytes, ...]:
+        return self.get(_encode_header_name(item))
 
     def keys(self) -> tuple[bytes, ...]:
         results = []
@@ -176,11 +235,11 @@ def contains(self, key: bytes) -> bool:
                 return True
         return False
 
-    def __delitem__(self, key: bytes):
-        self.remove(key)
+    def __delitem__(self, key: str | bytes):
+        self.remove(_encode_header_name(key))
 
-    def __contains__(self, key: bytes) -> bool:
-        return self.contains(key)
+    def __contains__(self, key: str | bytes) -> bool:
+        return self.contains(_encode_header_name(key))
 
     def __repr__(self) -> str:
         return f"<Headers {self.values!r}>"

blacksheep/headers.pyi

@@ -1,5 +1,7 @@
-from collections.abc import MutableSequence
-from typing import Generator
+from collections.abc import Iterable, Mapping, MutableSequence
+from typing import Generator, TypeVar
+
+_HeaderName = TypeVar("_HeaderName", bound=str | bytes)
 
 class Header:
     def __init__(self, name: bytes, value: bytes):
@@ -16,12 +18,28 @@ class Headers:
     def __init__(self, values: list[HeaderType] | None = None):
         self.values = values
 
+    @classmethod
+    def from_values(
+        cls,
+        values: Mapping[_HeaderName, str | bytes]
+        | Iterable[tuple[str | bytes, str | bytes]],
+    ) -> "Headers": ...
+    def add_text(self, name: str | bytes, value: str | bytes) -> None: ...
     def get(self, name: bytes) -> tuple[bytes]: ...
+    def get_text_values(
+        self, name: str | bytes, encoding: str = "latin-1"
+    ) -> tuple[str, ...]: ...
+    def get_first_text(
+        self, name: str | bytes, encoding: str = "latin-1"
+    ) -> str | None: ...
+    def get_single_text(
+        self, name: str | bytes, encoding: str = "latin-1"
+    ) -> str: ...
     def get_tuples(self, name: bytes) -> list[HeaderType]: ...
     def get_first(self, key: bytes) -> bytes | None: ...
     def get_single(self, key: bytes) -> bytes: ...
     def merge(self, values: list[HeaderType]) -> None: ...
-    def update(self, values: dict[bytes, bytes]) -> None: ...
+    def update(self, values: Mapping[_HeaderName, bytes | str]) -> None: ...
     def items(self) -> Generator[HeaderType, None, None]: ...
     def clone(self) -> "Headers": ...
     def add_many(
@@ -37,13 +55,13 @@ class Headers:
         self, other: "Headers" | Header | HeaderType | MutableSequence
     ) -> "Headers": ...
     def __iter__(self) -> Generator[HeaderType, None, None]: ...
-    def __setitem__(self, key: bytes, value: bytes) -> None: ...
-    def __getitem__(self, item: bytes) -> tuple[bytes]: ...
+    def __setitem__(self, key: str | bytes, value: str | bytes) -> None: ...
+    def __getitem__(self, item: str | bytes) -> tuple[bytes, ...]: ...
     def keys(self) -> tuple[bytes, ...]: ...
     def add(self, name: bytes, value: bytes) -> None: ...
     def set(self, name: bytes, value: bytes) -> None: ...
     def remove(self, key: bytes) -> None: ...
     def contains(self, key: bytes) -> bool: ...
-    def __delitem__(self, key: bytes) -> None: ...
-    def __contains__(self, key: bytes) -> bool: ...
+    def __delitem__(self, key: str | bytes) -> None: ...
+    def __contains__(self, key: str | bytes) -> bool: ...
     def __repr__(self) -> str: ...