Establish basic file browser and index fallback

.coveragerc

@@ -9,6 +9,7 @@ omit =
     sanic/simple.py
     sanic/utils.py
     sanic/cli
+    sanic/pages
 
 [html]
 directory = coverage

codecov.yml

@@ -17,7 +17,8 @@ ignore:
   - "sanic/compat.py"
   - "sanic/simple.py"
   - "sanic/utils.py"
-  - "sanic/cli"
+  - "sanic/cli/"
+  - "sanic/pages/"
   - ".github/"
   - "changelogs/"
   - "docker/"

pyproject.toml

@@ -23,5 +23,6 @@ module = [
     "trustme.*",
     "sanic_routing.*",
     "aioquic.*",
+    "html5tagger.*",
 ]
 ignore_missing_imports = true

sanic/app.py

@@ -72,14 +72,14 @@
 from sanic.middleware import Middleware, MiddlewareLocation
 from sanic.mixins.listeners import ListenerEvent
 from sanic.mixins.startup import StartupMixin
+from sanic.mixins.static import StaticHandleMixin
 from sanic.models.futures import (
     FutureException,
     FutureListener,
     FutureMiddleware,
     FutureRegistry,
     FutureRoute,
     FutureSignal,
-    FutureStatic,
 )
 from sanic.models.handler_types import ListenerType, MiddlewareType
 from sanic.models.handler_types import Sanic as SanicVar
@@ -106,7 +106,7 @@
     enable_windows_color_support()
 
 
-class Sanic(BaseSanic, StartupMixin, metaclass=TouchUpMeta):
+class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta):
     """
     The main application instance
     """
@@ -441,9 +441,6 @@ def _apply_route(self, route: FutureRoute) -> List[Route]:
 
         return routes
 
-    def _apply_static(self, static: FutureStatic) -> Route:
-        return self._register_static(static)
-
     def _apply_middleware(
         self,
         middleware: FutureMiddleware,
@@ -890,11 +887,11 @@ async def handle_request(self, request: Request):  # no cov
             Union[
                 BaseHTTPResponse,
                 Coroutine[Any, Any, Optional[BaseHTTPResponse]],
+                ResponseStream,
             ]
         ] = None
         run_middleware = True
         try:
-
             await self.dispatch(
                 "http.routing.before",
                 inline=True,
@@ -926,7 +923,6 @@ async def handle_request(self, request: Request):  # no cov
                 and request.stream.request_body
                 and not route.extra.ignore_body
             ):
-
                 if hasattr(handler, "is_stream"):
                     # Streaming handler: lift the size limit
                     request.stream.request_max_size = float("inf")
@@ -1000,7 +996,7 @@ async def handle_request(self, request: Request):  # no cov
                 ...
                 await response.send(end_stream=True)
             elif isinstance(response, ResponseStream):
-                resp = await response(request)  # type: ignore
+                resp = await response(request)
                 await self.dispatch(
                     "http.lifecycle.response",
                     inline=True,
@@ -1009,7 +1005,7 @@ async def handle_request(self, request: Request):  # no cov
                         "response": resp,
                     },
                 )
-                await response.eof()  # type: ignore
+                await response.eof()
             else:
                 if not hasattr(handler, "is_websocket"):
                     raise ServerError(

sanic/application/logo.py

@@ -40,6 +40,8 @@
 
 """  # noqa
 
+SVG_LOGO = """<svg id=logo alt=Sanic viewBox="0 0 964 279"><path d="M107 222c9-2 10-20 1-22s-20-2-30-2-17 7-16 14 6 10 15 10h30zm115-1c16-2 30-11 35-23s6-24 2-33-6-14-15-20-24-11-38-10c-7 3-10 13-5 19s17-1 24 4 15 14 13 24-5 15-14 18-50 0-74 0h-17c-6 4-10 15-4 20s16 2 23 3zM251 83q9-1 9-7 0-15-10-16h-13c-10 6-10 20 0 22zM147 60c-4 0-10 3-11 11s5 13 10 12 42 0 67 0c5-3 7-10 6-15s-4-8-9-8zm-33 1c-8 0-16 0-24 3s-20 10-25 20-6 24-4 36 15 22 26 27 78 8 94 3c4-4 4-12 0-18s-69 8-93-10c-8-7-9-23 0-30s12-10 20-10 12 2 16-3 1-15-5-18z" fill="#ff0d68"/><path d="M676 74c0-14-18-9-20 0s0 30 0 39 20 9 20 2zm-297-10c-12 2-15 12-23 23l-41 58H340l22-30c8-12 23-13 30-4s20 24 24 38-10 10-17 10l-68 2q-17 1-48 30c-7 6-10 20 0 24s15-8 20-13 20 -20 58-21h50 c20 2 33 9 52 30 8 10 24-4 16-13L384 65q-3-2-5-1zm131 0c-10 1-12 12-11 20v96c1 10-3 23 5 32s20-5 17-15c0-23-3-46 2-67 5-12 22-14 32-5l103 87c7 5 19 1 18-9v-64c-3-10-20-9-21 2s-20 22-30 13l-97-80c-5-4-10-10-18-10zM701 76v128c2 10 15 12 20 4s0-102 0-124s-20-18-20-7z    M850 63c-35 0-69-2-86 15s-20 60-13 66 13 8 16 0 1-10 1-27 12-26 20-32 66-5 85-5 31 4 31-10-18-7-54-7M764 159c-6-2-15-2-16 12s19 37 33 43 23 8 25-4-4-11-11-14q-9-3-22-18c-4-7-3-16-10-19zM828 196c-4 0-8 1-10 5s-4 12 0 15 8 2 12 2h60c5 0 10-2 12-6 3-7-1-16-8-16z" fill="#e1e1e1"/></svg>"""  # noqa
+
 ansi_pattern = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
 
 

sanic/base/root.py

@@ -9,13 +9,15 @@
 from sanic.mixins.middleware import MiddlewareMixin
 from sanic.mixins.routes import RouteMixin
 from sanic.mixins.signals import SignalMixin
+from sanic.mixins.static import StaticMixin
 
 
 VALID_NAME = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_\-]*$")
 
 
 class BaseSanic(
     RouteMixin,
+    StaticMixin,
     MiddlewareMixin,
     ListenerMixin,
     ExceptionMixin,

sanic/constants.py

@@ -4,7 +4,6 @@
 
 
 class HTTPMethod(UpperStrEnum):
-
     GET = auto()
     POST = auto()
     PUT = auto()
@@ -15,7 +14,6 @@ class HTTPMethod(UpperStrEnum):
 
 
 class LocalCertCreator(UpperStrEnum):
-
     AUTO = auto()
     TRUSTME = auto()
     MKCERT = auto()

sanic/errorpages.py

@@ -12,6 +12,7 @@
 will attempt to provide an appropriate response format based upon the
 request type.
 """
+from __future__ import annotations
 
 import sys
 import typing as t
@@ -21,8 +22,7 @@
 
 from sanic.exceptions import BadRequest, SanicException
 from sanic.helpers import STATUS_CODES
-from sanic.request import Request
-from sanic.response import HTTPResponse, html, json, text
+from sanic.response import html, json, text
 
 
 dumps: t.Callable[..., str]
@@ -33,6 +33,8 @@
 except ImportError:  # noqa
     from json import dumps
 
+if t.TYPE_CHECKING:
+    from sanic import HTTPResponse, Request
 
 DEFAULT_FORMAT = "auto"
 FALLBACK_TEXT = (

sanic/handlers/__init__.py

@@ -0,0 +1,10 @@
+from .content_range import ContentRangeHandler
+from .directory import DirectoryHandler
+from .error import ErrorHandler
+
+
+__all__ = (
+    "ContentRangeHandler",
+    "DirectoryHandler",
+    "ErrorHandler",
+)

sanic/handlers/content_range.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from sanic.exceptions import (
+    HeaderNotFound,
+    InvalidRangeType,
+    RangeNotSatisfiable,
+)
+
+
+class ContentRangeHandler:
+    """
+    A mechanism to parse and process the incoming request headers to
+    extract the content range information.
+
+    :param request: Incoming api request
+    :param stats: Stats related to the content
+
+    :type request: :class:`sanic.request.Request`
+    :type stats: :class:`posix.stat_result`
+
+    :ivar start: Content Range start
+    :ivar end: Content Range end
+    :ivar size: Length of the content
+    :ivar total: Total size identified by the :class:`posix.stat_result`
+        instance
+    :ivar ContentRangeHandler.headers: Content range header ``dict``
+    """
+
+    __slots__ = ("start", "end", "size", "total", "headers")
+
+    def __init__(self, request, stats):
+        self.total = stats.st_size
+        _range = request.headers.getone("range", None)
+        if _range is None:
+            raise HeaderNotFound("Range Header Not Found")
+        unit, _, value = tuple(map(str.strip, _range.partition("=")))
+        if unit != "bytes":
+            raise InvalidRangeType(
+                "%s is not a valid Range Type" % (unit,), self
+            )
+        start_b, _, end_b = tuple(map(str.strip, value.partition("-")))
+        try:
+            self.start = int(start_b) if start_b else None
+        except ValueError:
+            raise RangeNotSatisfiable(
+                "'%s' is invalid for Content Range" % (start_b,), self
+            )
+        try:
+            self.end = int(end_b) if end_b else None
+        except ValueError:
+            raise RangeNotSatisfiable(
+                "'%s' is invalid for Content Range" % (end_b,), self
+            )
+        if self.end is None:
+            if self.start is None:
+                raise RangeNotSatisfiable(
+                    "Invalid for Content Range parameters", self
+                )
+            else:
+                # this case represents `Content-Range: bytes 5-`
+                self.end = self.total - 1
+        else:
+            if self.start is None:
+                # this case represents `Content-Range: bytes -5`
+                self.start = self.total - self.end
+                self.end = self.total - 1
+        if self.start >= self.end:
+            raise RangeNotSatisfiable(
+                "Invalid for Content Range parameters", self
+            )
+        self.size = self.end - self.start + 1
+        self.headers = {
+            "Content-Range": "bytes %s-%s/%s"
+            % (self.start, self.end, self.total)
+        }
+
+    def __bool__(self):
+        return self.size > 0

sanic/handlers/directory.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from datetime import datetime
+from operator import itemgetter
+from pathlib import Path
+from stat import S_ISDIR
+from typing import Dict, Iterable, Optional, Sequence, Union, cast
+
+from sanic.exceptions import NotFound
+from sanic.pages.directory_page import DirectoryPage, FileInfo
+from sanic.request import Request
+from sanic.response import file, html, redirect
+
+
+class DirectoryHandler:
+    def __init__(
+        self,
+        uri: str,
+        directory: Path,
+        directory_view: bool = False,
+        index: Optional[Union[str, Sequence[str]]] = None,
+    ) -> None:
+        if isinstance(index, str):
+            index = [index]
+        elif index is None:
+            index = []
+        self.base = uri.strip("/")
+        self.directory = directory
+        self.directory_view = directory_view
+        self.index = tuple(index)
+
+    async def handle(self, request: Request, path: str):
+        current = path.strip("/")[len(self.base) :].strip("/")  # noqa: E203
+        for file_name in self.index:
+            index_file = self.directory / current / file_name
+            if index_file.is_file():
+                return await file(index_file)
+
+        if self.directory_view:
+            return self._index(
+                self.directory / current, path, request.app.debug
+            )
+
+        if self.index:
+            raise NotFound("File not found")
+
+        raise IsADirectoryError(f"{self.directory.as_posix()} is a directory")
+
+    def _index(self, location: Path, path: str, debug: bool):
+        # Remove empty path elements, append slash
+        if "//" in path or not path.endswith("/"):
+            return redirect(
+                "/" + "".join([f"{p}/" for p in path.split("/") if p])
+            )
+
+        # Render file browser
+        page = DirectoryPage(self._iter_files(location), path, debug)
+        return html(page.render())
+
+    def _prepare_file(self, path: Path) -> Dict[str, Union[int, str]]:
+        stat = path.stat()
+        modified = (
+            datetime.fromtimestamp(stat.st_mtime)
+            .isoformat()[:19]
+            .replace("T", " ")
+        )
+        is_dir = S_ISDIR(stat.st_mode)
+        icon = "📁" if is_dir else "📄"
+        file_name = path.name
+        if is_dir:
+            file_name += "/"
+        return {
+            "priority": is_dir * -1,
+            "file_name": file_name,
+            "icon": icon,
+            "file_access": modified,
+            "file_size": stat.st_size,
+        }
+
+    def _iter_files(self, location: Path) -> Iterable[FileInfo]:
+        prepared = [self._prepare_file(f) for f in location.iterdir()]
+        for item in sorted(prepared, key=itemgetter("priority", "file_name")):
+            del item["priority"]
+            yield cast(FileInfo, item)