Support FileResponse from any pathlib.Path compatible object

aiohttp/typedefs.py

@@ -1,14 +1,17 @@
 import json
 import os
 from typing import (
+    IO,
     TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
     Iterable,
     Mapping,
+    Protocol,
     Tuple,
     Union,
+    runtime_checkable,
 )
 
 from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy, istr
@@ -52,3 +55,24 @@
 Middleware = Callable[["Request", Handler], Awaitable["StreamResponse"]]
 
 PathLike = Union[str, "os.PathLike[str]"]
+
+
+class PathlibPathNamedLike(Protocol):
+    def is_file(self) -> bool:
+        ...
+
+
+@runtime_checkable
+class PathlibPathLike(Protocol):
+    """pathlib.Path interface used by aiohttp."""
+
+    name: str
+
+    def open(self, mode: str) -> IO[Any]:
+        ...
+
+    def stat(self, *, follow_symlinks=True) -> os.stat_result:
+        ...
+
+    def with_name(self, name: str) -> PathlibPathNamedLike:
+        ...

aiohttp/web_fileresponse.py

@@ -11,13 +11,14 @@
     Final,
     Optional,
     Tuple,
+    Union,
     cast,
 )
 
 from . import hdrs
 from .abc import AbstractStreamWriter
 from .helpers import ETAG_ANY, ETag, must_be_empty_body
-from .typedefs import LooseHeaders, PathLike
+from .typedefs import LooseHeaders, PathlibPathLike, PathLike
 from .web_exceptions import (
     HTTPNotModified,
     HTTPPartialContent,
@@ -43,15 +44,20 @@ class FileResponse(StreamResponse):
 
     def __init__(
         self,
-        path: PathLike,
+        path: Union[PathLike, PathlibPathLike],
         chunk_size: int = 256 * 1024,
         status: int = 200,
         reason: Optional[str] = None,
         headers: Optional[LooseHeaders] = None,
     ) -> None:
         super().__init__(status=status, reason=reason, headers=headers)
 
-        self._path = pathlib.Path(path)
+        if isinstance(path, str) or (
+            not isinstance(path, PathlibPathLike) and isinstance(path, os.PathLike)
+        ):
+            path = pathlib.Path(path)
+
+        self._path = path
         self._chunk_size = chunk_size
 
     async def _sendfile_fallback(

tests/test_web_sendfile.py

@@ -1,3 +1,5 @@
+from io import BytesIO
+from os import stat_result
 from pathlib import Path
 from typing import Any
 from unittest import mock
@@ -114,3 +116,56 @@ def test_status_controlled_by_user(loop: Any) -> None:
     loop.run_until_complete(file_sender.prepare(request))
 
     assert file_sender._status == 203
+
+
+def test_custom_path(loop: Any) -> None:
+    request = make_mocked_request("GET", "http://python.org/hello")
+
+    # ZipFile has no with_name and stat
+    # file = BytesIO()
+    # zipfile = ZipFile(file, "w")
+    # zipfile.writestr("hello", "world")
+    # filepath = ZipPath(zipfile)
+
+    class MyPath:
+        name = "hello"
+        content = b"world"
+
+        def open(self, mode: str, *args, **kwargs):
+            return BytesIO(self.content)
+
+        def stat(self, **_):
+            ts = 1701435976
+            return stat_result(
+                (
+                    0o444,
+                    -1,
+                    -1,
+                    1,
+                    0,
+                    0,
+                    len(self.content),
+                    ts,
+                    ts,
+                    ts,
+                    ts,
+                    ts,
+                    ts,
+                    ts * 1000000000,
+                    ts * 1000000000,
+                    ts * 1000000000,
+                )
+            )
+
+        def with_name(self, name):
+            return NoPath()
+
+    class NoPath:
+        def is_file(self):
+            return False
+
+    filepath = MyPath()
+    file_sender = FileResponse(filepath)
+    file_sender._sendfile = make_mocked_coro(None)  # type: ignore[method-assign]
+
+    loop.run_until_complete(file_sender.prepare(request))