Enable mypy option --disallow-any-generics

httpcore/_api.py

@@ -1,17 +1,17 @@
 from contextlib import contextmanager
 from typing import Iterator, Optional, Union
 
-from ._models import URL, Response
+from ._models import URL, Extensions, HeaderTypes, Response
 from ._sync.connection_pool import ConnectionPool
 
 
 def request(
     method: Union[bytes, str],
     url: Union[URL, bytes, str],
     *,
-    headers: Union[dict, list, None] = None,
+    headers: HeaderTypes = None,
     content: Union[bytes, Iterator[bytes], None] = None,
-    extensions: Optional[dict] = None,
+    extensions: Optional[Extensions] = None,
 ) -> Response:
     """
     Sends an HTTP request, returning the response.
@@ -50,9 +50,9 @@ def stream(
     method: Union[bytes, str],
     url: Union[URL, bytes, str],
     *,
-    headers: Union[dict, list, None] = None,
+    headers: HeaderTypes = None,
     content: Union[bytes, Iterator[bytes], None] = None,
-    extensions: Optional[dict] = None,
+    extensions: Optional[Extensions] = None,
 ) -> Iterator[Response]:
     """
     Sends an HTTP request, returning the response within a content manager.

httpcore/_async/interfaces.py

@@ -3,6 +3,8 @@
 
 from .._models import (
     URL,
+    Extensions,
+    HeaderTypes,
     Origin,
     Request,
     Response,
@@ -19,9 +21,9 @@ async def request(
         method: Union[bytes, str],
         url: Union[URL, bytes, str],
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, AsyncIterator[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> Response:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")
@@ -51,9 +53,9 @@ async def stream(
         method: Union[bytes, str],
         url: Union[URL, bytes, str],
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, AsyncIterator[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> AsyncIterator[Response]:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")

httpcore/_exceptions.py

@@ -1,9 +1,11 @@
 import contextlib
-from typing import Dict, Iterator, Type
+from typing import Iterator, Mapping, Type
+
+ExceptionMapping = Mapping[Type[Exception], Type[Exception]]
 
 
 @contextlib.contextmanager
-def map_exceptions(map: Dict[Type, Type]) -> Iterator[None]:
+def map_exceptions(map: ExceptionMapping) -> Iterator[None]:
     try:
         yield
     except Exception as exc:  # noqa: PIE786

httpcore/_models.py

@@ -18,6 +18,9 @@
 
 HeadersAsSequence = Sequence[Tuple[Union[bytes, str], Union[bytes, str]]]
 HeadersAsMapping = Mapping[Union[bytes, str], Union[bytes, str]]
+HeaderTypes = Union[HeadersAsSequence, HeadersAsMapping, None]
+
+Extensions = Mapping[str, Any]
 
 
 def enforce_bytes(value: Union[bytes, str], *, name: str) -> bytes:
@@ -318,9 +321,9 @@ def __init__(
         method: Union[bytes, str],
         url: Union[URL, bytes, str],
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> None:
         """
         Parameters:
@@ -356,9 +359,9 @@ def __init__(
         self,
         status: int,
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, Iterable[bytes], AsyncIterable[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> None:
         """
         Parameters:
@@ -376,7 +379,7 @@ def __init__(
         self.stream: Union[Iterable[bytes], AsyncIterable[bytes]] = enforce_stream(
             content, name="content"
         )
-        self.extensions: dict = {} if extensions is None else extensions
+        self.extensions = {} if extensions is None else extensions
 
         self._stream_consumed = False
 

httpcore/_sync/interfaces.py

@@ -3,6 +3,8 @@
 
 from .._models import (
     URL,
+    Extensions,
+    HeaderTypes,
     Origin,
     Request,
     Response,
@@ -19,9 +21,9 @@ def request(
         method: Union[bytes, str],
         url: Union[URL, bytes, str],
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, Iterator[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> Response:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")
@@ -51,9 +53,9 @@ def stream(
         method: Union[bytes, str],
         url: Union[URL, bytes, str],
         *,
-        headers: Union[dict, list, None] = None,
+        headers: HeaderTypes = None,
         content: Union[bytes, Iterator[bytes], None] = None,
-        extensions: Optional[dict] = None,
+        extensions: Optional[Extensions] = None,
     ) -> Iterator[Response]:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")

httpcore/_synchronization.py

@@ -4,7 +4,7 @@
 
 import anyio
 
-from ._exceptions import PoolTimeout, map_exceptions
+from ._exceptions import ExceptionMapping, PoolTimeout, map_exceptions
 
 
 class AsyncLock:
@@ -32,7 +32,7 @@ def set(self) -> None:
         self._event.set()
 
     async def wait(self, timeout: Optional[float] = None) -> None:
-        exc_map: dict = {TimeoutError: PoolTimeout}
+        exc_map: ExceptionMapping = {TimeoutError: PoolTimeout}
         with map_exceptions(exc_map):
             with anyio.fail_after(timeout):
                 await self._event.wait()

httpcore/_trace.py

@@ -1,12 +1,12 @@
 from types import TracebackType
-from typing import Any, Optional, Type
+from typing import Any, Dict, Optional, Type
 
 from ._models import Request
 
 
 class Trace:
     def __init__(
-        self, name: str, request: Request, kwargs: Optional[dict] = None
+        self, name: str, request: Request, kwargs: Optional[Dict[str, Any]] = None
     ) -> None:
         self.name = name
         self.trace = request.extensions.get("trace")
@@ -27,7 +27,7 @@ def __exit__(
     ) -> None:
         if self.trace is not None:
             if exc_value is None:
-                info: dict = {"return_value": self.return_value}
+                info = {"return_value": self.return_value}
                 self.trace(f"{self.name}.complete", info)
             else:
                 info = {"exception": exc_value}
@@ -47,7 +47,7 @@ async def __aexit__(
     ) -> None:
         if self.trace is not None:
             if exc_value is None:
-                info: dict = {"return_value": self.return_value}
+                info = {"return_value": self.return_value}
                 await self.trace(f"{self.name}.complete", info)
             else:
                 info = {"exception": exc_value}

httpcore/backends/sync.py

@@ -5,6 +5,7 @@
 from .._exceptions import (
     ConnectError,
     ConnectTimeout,
+    ExceptionMapping,
     ReadError,
     ReadTimeout,
     WriteError,
@@ -20,7 +21,7 @@ def __init__(self, sock: socket.socket) -> None:
         self._sock = sock
 
     def read(self, max_bytes: int, timeout: typing.Optional[float] = None) -> bytes:
-        exc_map = {socket.timeout: ReadTimeout, OSError: ReadError}
+        exc_map: ExceptionMapping = {socket.timeout: ReadTimeout, OSError: ReadError}
         with map_exceptions(exc_map):
             self._sock.settimeout(timeout)
             return self._sock.recv(max_bytes)
@@ -29,7 +30,7 @@ def write(self, buffer: bytes, timeout: typing.Optional[float] = None) -> None:
         if not buffer:
             return
 
-        exc_map = {socket.timeout: WriteTimeout, OSError: WriteError}
+        exc_map: ExceptionMapping = {socket.timeout: WriteTimeout, OSError: WriteError}
         with map_exceptions(exc_map):
             while buffer:
                 self._sock.settimeout(timeout)
@@ -45,7 +46,10 @@ def start_tls(
         server_hostname: typing.Optional[str] = None,
         timeout: typing.Optional[float] = None,
     ) -> NetworkStream:
-        exc_map = {socket.timeout: ConnectTimeout, OSError: ConnectError}
+        exc_map: ExceptionMapping = {
+            socket.timeout: ConnectTimeout,
+            OSError: ConnectError,
+        }
         with map_exceptions(exc_map):
             try:
                 self._sock.settimeout(timeout)
@@ -81,7 +85,10 @@ def connect_tcp(
     ) -> NetworkStream:
         address = (host, port)
         source_address = None if local_address is None else (local_address, 0)
-        exc_map = {socket.timeout: ConnectTimeout, OSError: ConnectError}
+        exc_map: ExceptionMapping = {
+            socket.timeout: ConnectTimeout,
+            OSError: ConnectError,
+        }
         with map_exceptions(exc_map):
             sock = socket.create_connection(
                 address, timeout, source_address=source_address
@@ -91,7 +98,10 @@ def connect_tcp(
     def connect_unix_socket(
         self, path: str, timeout: typing.Optional[float] = None
     ) -> NetworkStream:  # pragma: nocover
-        exc_map = {socket.timeout: ConnectTimeout, OSError: ConnectError}
+        exc_map: ExceptionMapping = {
+            socket.timeout: ConnectTimeout,
+            OSError: ConnectError,
+        }
         with map_exceptions(exc_map):
             sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
             sock.settimeout(timeout)

httpcore/backends/trio.py

@@ -6,6 +6,7 @@
 from .._exceptions import (
     ConnectError,
     ConnectTimeout,
+    ExceptionMapping,
     ReadError,
     ReadTimeout,
     WriteError,
@@ -23,7 +24,10 @@ async def read(
         self, max_bytes: int, timeout: typing.Optional[float] = None
     ) -> bytes:
         timeout_or_inf = float("inf") if timeout is None else timeout
-        exc_map = {trio.TooSlowError: ReadTimeout, trio.BrokenResourceError: ReadError}
+        exc_map: ExceptionMapping = {
+            trio.TooSlowError: ReadTimeout,
+            trio.BrokenResourceError: ReadError,
+        }
         with map_exceptions(exc_map):
             with trio.fail_after(timeout_or_inf):
                 data: bytes = await self._stream.receive_some(max_bytes=max_bytes)
@@ -36,7 +40,7 @@ async def write(
             return
 
         timeout_or_inf = float("inf") if timeout is None else timeout
-        exc_map = {
+        exc_map: ExceptionMapping = {
             trio.TooSlowError: WriteTimeout,
             trio.BrokenResourceError: WriteError,
         }
@@ -54,7 +58,7 @@ async def start_tls(
         timeout: typing.Optional[float] = None,
     ) -> AsyncNetworkStream:
         timeout_or_inf = float("inf") if timeout is None else timeout
-        exc_map = {
+        exc_map: ExceptionMapping = {
             trio.TooSlowError: ConnectTimeout,
             trio.BrokenResourceError: ConnectError,
         }
@@ -109,15 +113,15 @@ async def connect_tcp(
         local_address: typing.Optional[str] = None,
     ) -> AsyncNetworkStream:
         timeout_or_inf = float("inf") if timeout is None else timeout
-        exc_map = {
+        exc_map: ExceptionMapping = {
             trio.TooSlowError: ConnectTimeout,
             trio.BrokenResourceError: ConnectError,
             OSError: ConnectError,
         }
         # Trio supports 'local_address' from 0.16.1 onwards.
         # We only include the keyword argument if a local_address
         # argument has been passed.
-        kwargs: dict = {} if local_address is None else {"local_address": local_address}
+        kwargs = {} if local_address is None else {"local_address": local_address}
         with map_exceptions(exc_map):
             with trio.fail_after(timeout_or_inf):
                 stream: trio.abc.Stream = await trio.open_tcp_stream(
@@ -129,7 +133,7 @@ async def connect_unix_socket(
         self, path: str, timeout: typing.Optional[float] = None
     ) -> AsyncNetworkStream:  # pragma: nocover
         timeout_or_inf = float("inf") if timeout is None else timeout
-        exc_map = {
+        exc_map: ExceptionMapping = {
             trio.TooSlowError: ConnectTimeout,
             trio.BrokenResourceError: ConnectError,
             OSError: ConnectError,

setup.cfg

@@ -11,6 +11,7 @@ show_error_codes = True
 warn_unused_ignores = True
 disallow_untyped_calls = True
 warn_return_any = True
+disallow_any_generics = True
 
 [mypy-tests.*]
 disallow_untyped_defs = False