Use PathLike (#6876)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: Dreamsorcerer

aiohttp/web.py

@@ -1,11 +1,11 @@
 import asyncio
 import logging
+import os
 import socket
 import sys
 from argparse import ArgumentParser
 from collections.abc import Iterable
 from importlib import import_module
-from pathlib import Path
 from typing import (
     Any,
     Awaitable,
@@ -22,6 +22,7 @@
 from .abc import AbstractAccessLogger
 from .helpers import AppKey as AppKey
 from .log import access_logger
+from .typedefs import PathLike
 from .web_app import Application as Application, CleanupError as CleanupError
 from .web_exceptions import (
     HTTPAccepted as HTTPAccepted,
@@ -293,7 +294,7 @@ async def _run_app(
     *,
     host: Optional[Union[str, HostSequence]] = None,
     port: Optional[int] = None,
-    path: Optional[Union[str, Path]] = None,
+    path: Union[PathLike, TypingIterable[PathLike], None] = None,
     sock: Optional[Union[socket.socket, TypingIterable[socket.socket]]] = None,
     shutdown_timeout: float = 60.0,
     keepalive_timeout: float = 75.0,
@@ -369,7 +370,7 @@ async def _run_app(
             )
 
         if path is not None:
-            if isinstance(path, (str, bytes, bytearray, memoryview, Path)):
+            if isinstance(path, (str, os.PathLike)):
                 sites.append(
                     UnixSite(
                         runner,
@@ -467,7 +468,7 @@ def run_app(
     debug: bool = False,
     host: Optional[Union[str, HostSequence]] = None,
     port: Optional[int] = None,
-    path: Optional[Union[str, Path]] = None,
+    path: Union[PathLike, TypingIterable[PathLike], None] = None,
     sock: Optional[Union[socket.socket, TypingIterable[socket.socket]]] = None,
     shutdown_timeout: float = 60.0,
     keepalive_timeout: float = 75.0,

aiohttp/web_fileresponse.py

@@ -21,7 +21,7 @@
 from . import hdrs
 from .abc import AbstractStreamWriter
 from .helpers import ETAG_ANY, ETag
-from .typedefs import LooseHeaders
+from .typedefs import LooseHeaders, PathLike
 from .web_exceptions import (
     HTTPNotModified,
     HTTPPartialContent,
@@ -47,18 +47,15 @@ class FileResponse(StreamResponse):
 
     def __init__(
         self,
-        path: Union[str, pathlib.Path],
+        path: PathLike,
         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)
 
-        if isinstance(path, str):
-            path = pathlib.Path(path)
-
-        self._path = path
+        self._path = pathlib.Path(path)
         self._chunk_size = chunk_size
 
     async def _sendfile_fallback(

aiohttp/web_runner.py

@@ -2,14 +2,14 @@
 import signal
 import socket
 from abc import ABC, abstractmethod
-from pathlib import Path
-from typing import Any, List, Optional, Set, Type, Union
+from typing import Any, List, Optional, Set, Type
 
 from yarl import URL
 
 from .abc import AbstractAccessLogger, AbstractStreamWriter
 from .http_parser import RawRequestMessage
 from .streams import StreamReader
+from .typedefs import PathLike
 from .web_app import Application
 from .web_log import AccessLogger
 from .web_protocol import RequestHandler
@@ -142,7 +142,7 @@ class UnixSite(BaseSite):
     def __init__(
         self,
         runner: "BaseRunner",
-        path: Union[str, Path],
+        path: PathLike,
         *,
         shutdown_timeout: float = 60.0,
         ssl_context: Optional[SSLContext] = None,

aiohttp/web_urldispatcher.py

@@ -559,14 +559,12 @@ def __init__(
     def url_for(  # type: ignore[override]
         self,
         *,
-        filename: Union[str, Path],
+        filename: PathLike,
         append_version: Optional[bool] = None,
     ) -> URL:
         if append_version is None:
             append_version = self._append_version
-        if isinstance(filename, Path):
-            filename = str(filename)
-        filename = filename.lstrip("/")
+        filename = str(filename).lstrip("/")
 
         url = URL.build(path=self._prefix, encoded=True)
         # filename is not encoded

docs/web_reference.rst

@@ -2846,7 +2846,8 @@ Utilities
    :param path: file system path for HTTP server Unix domain socket.
                     A sequence of file system paths can be used to bind
                     multiple domain sockets. Listening on Unix domain
-                    sockets is not supported by all operating systems, :class:`str` or :class:`pathlib.Path` .
+                    sockets is not supported by all operating systems,
+                    :class:`str`, :class:`pathlib.Path` or an iterable of these.
 
    :param socket.socket sock: a preexisting socket object to accept connections on.
                        A sequence of socket objects can be passed.

tests/test_web_sendfile.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from typing import Any
 from unittest import mock
 
@@ -11,19 +12,17 @@ def test_using_gzip_if_header_present_and_file_available(loop: Any) -> None:
         "GET", "http://python.org/logo.png", headers={hdrs.ACCEPT_ENCODING: "gzip"}
     )
 
-    gz_filepath = mock.Mock()
-    gz_filepath.open = mock.mock_open()
+    gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.is_file.return_value = True
-    gz_filepath.stat.return_value = mock.MagicMock()
     gz_filepath.stat.return_value.st_size = 1024
     gz_filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
-    filepath = mock.Mock()
+    filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.open = mock.mock_open()
     filepath.with_name.return_value = gz_filepath
 
     file_sender = FileResponse(filepath)
+    file_sender._path = filepath
     file_sender._sendfile = make_mocked_coro(None)  # type: ignore[assignment]
 
     loop.run_until_complete(file_sender.prepare(request))
@@ -35,19 +34,17 @@ def test_using_gzip_if_header_present_and_file_available(loop: Any) -> None:
 def test_gzip_if_header_not_present_and_file_available(loop: Any) -> None:
     request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
 
-    gz_filepath = mock.Mock()
-    gz_filepath.open = mock.mock_open()
+    gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.is_file.return_value = True
 
-    filepath = mock.Mock()
+    filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.open = mock.mock_open()
     filepath.with_name.return_value = gz_filepath
-    filepath.stat.return_value = mock.MagicMock()
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
     file_sender = FileResponse(filepath)
+    file_sender._path = filepath
     file_sender._sendfile = make_mocked_coro(None)  # type: ignore[assignment]
 
     loop.run_until_complete(file_sender.prepare(request))
@@ -59,19 +56,17 @@ def test_gzip_if_header_not_present_and_file_available(loop: Any) -> None:
 def test_gzip_if_header_not_present_and_file_not_available(loop: Any) -> None:
     request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
 
-    gz_filepath = mock.Mock()
-    gz_filepath.open = mock.mock_open()
+    gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.is_file.return_value = False
 
-    filepath = mock.Mock()
+    filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.open = mock.mock_open()
     filepath.with_name.return_value = gz_filepath
-    filepath.stat.return_value = mock.MagicMock()
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
     file_sender = FileResponse(filepath)
+    file_sender._path = filepath
     file_sender._sendfile = make_mocked_coro(None)  # type: ignore[assignment]
 
     loop.run_until_complete(file_sender.prepare(request))
@@ -85,19 +80,17 @@ def test_gzip_if_header_present_and_file_not_available(loop: Any) -> None:
         "GET", "http://python.org/logo.png", headers={hdrs.ACCEPT_ENCODING: "gzip"}
     )
 
-    gz_filepath = mock.Mock()
-    gz_filepath.open = mock.mock_open()
+    gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.is_file.return_value = False
 
-    filepath = mock.Mock()
+    filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.open = mock.mock_open()
     filepath.with_name.return_value = gz_filepath
-    filepath.stat.return_value = mock.MagicMock()
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
     file_sender = FileResponse(filepath)
+    file_sender._path = filepath
     file_sender._sendfile = make_mocked_coro(None)  # type: ignore[assignment]
 
     loop.run_until_complete(file_sender.prepare(request))
@@ -109,14 +102,13 @@ def test_gzip_if_header_present_and_file_not_available(loop: Any) -> None:
 def test_status_controlled_by_user(loop: Any) -> None:
     request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
 
-    filepath = mock.Mock()
+    filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
-    filepath.open = mock.mock_open()
-    filepath.stat.return_value = mock.MagicMock()
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
 
     file_sender = FileResponse(filepath, status=203)
+    file_sender._path = filepath
     file_sender._sendfile = make_mocked_coro(None)  # type: ignore[assignment]
 
     loop.run_until_complete(file_sender.prepare(request))