from __future__ import annotations
import sys
-from collections.abc import Iterator
-from typing import Any, Protocol
+from collections.abc import Awaitable, Iterator
+from typing import Any, Callable, Protocol
if sys.version_info >= (3, 10): # pragma: no cover
from typing import ParamSpec
else: # pragma: no cover
from typing_extensions import ParamSpec
-from starlette.types import ASGIApp
P = ParamSpec("P")
+_Scope = Any
+_Receive = Callable[[], Awaitable[Any]]
+_Send = Callable[[Any], Awaitable[None]]
+# Since `starlette.types.ASGIApp` type differs from `ASGIApplication` from `asgiref`
+# we need to define a more permissive version of ASGIApp that doesn't cause type errors.
+_ASGIApp = Callable[[_Scope, _Receive, _Send], Awaitable[None]]
+
+
class _MiddlewareFactory(Protocol[P]):
- def __call__(self, app: ASGIApp, /, *args: P.args, **kwargs: P.kwargs) -> ASGIApp: ... # pragma: no cover
+ def __call__(self, app: _ASGIApp, /, *args: P.args, **kwargs: P.kwargs) -> _ASGIApp: ... # pragma: no cover
class Middleware:
- def __init__(
- self,
- cls: _MiddlewareFactory[P],
- *args: P.args,
- **kwargs: P.kwargs,
- ) -> None:
+ def __init__(self, cls: _MiddlewareFactory[P], *args: P.args, **kwargs: P.kwargs) -> None:
self.cls = cls
self.args = args
self.kwargs = kwargs