Compare commits
	
		
			1 Commits
		
	
	
		
			accept-enh
			...
			22.12LTS
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
|   | fc82b2334b | 
| @@ -9,7 +9,6 @@ omit = | ||||
|     sanic/simple.py | ||||
|     sanic/utils.py | ||||
|     sanic/cli | ||||
|     sanic/pages | ||||
|  | ||||
| [html] | ||||
| directory = coverage | ||||
|   | ||||
| @@ -17,8 +17,7 @@ ignore: | ||||
|   - "sanic/compat.py" | ||||
|   - "sanic/simple.py" | ||||
|   - "sanic/utils.py" | ||||
|   - "sanic/cli/" | ||||
|   - "sanic/pages/" | ||||
|   - "sanic/cli" | ||||
|   - ".github/" | ||||
|   - "changelogs/" | ||||
|   - "docker/" | ||||
|   | ||||
| @@ -23,6 +23,5 @@ module = [ | ||||
|     "trustme.*", | ||||
|     "sanic_routing.*", | ||||
|     "aioquic.*", | ||||
|     "html5tagger.*", | ||||
| ] | ||||
| ignore_missing_imports = true | ||||
|   | ||||
							
								
								
									
										12
									
								
								sanic/app.py
									
									
									
									
									
								
							
							
						
						
									
										12
									
								
								sanic/app.py
									
									
									
									
									
								
							| @@ -72,7 +72,6 @@ from sanic.log import ( | ||||
| 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, | ||||
| @@ -80,6 +79,7 @@ from sanic.models.futures import ( | ||||
|     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 @@ if OS_IS_WINDOWS:  # no cov | ||||
|     enable_windows_color_support() | ||||
|  | ||||
|  | ||||
| class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
| class Sanic(BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
|     """ | ||||
|     The main application instance | ||||
|     """ | ||||
| @@ -441,6 +441,9 @@ class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
|  | ||||
|         return routes | ||||
|  | ||||
|     def _apply_static(self, static: FutureStatic) -> Route: | ||||
|         return self._register_static(static) | ||||
|  | ||||
|     def _apply_middleware( | ||||
|         self, | ||||
|         middleware: FutureMiddleware, | ||||
| @@ -887,7 +890,6 @@ class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
|             Union[ | ||||
|                 BaseHTTPResponse, | ||||
|                 Coroutine[Any, Any, Optional[BaseHTTPResponse]], | ||||
|                 ResponseStream, | ||||
|             ] | ||||
|         ] = None | ||||
|         run_middleware = True | ||||
| @@ -996,7 +998,7 @@ class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
|                 ... | ||||
|                 await response.send(end_stream=True) | ||||
|             elif isinstance(response, ResponseStream): | ||||
|                 resp = await response(request) | ||||
|                 resp = await response(request)  # type: ignore | ||||
|                 await self.dispatch( | ||||
|                     "http.lifecycle.response", | ||||
|                     inline=True, | ||||
| @@ -1005,7 +1007,7 @@ class Sanic(StaticHandleMixin, BaseSanic, StartupMixin, metaclass=TouchUpMeta): | ||||
|                         "response": resp, | ||||
|                     }, | ||||
|                 ) | ||||
|                 await response.eof() | ||||
|                 await response.eof()  # type: ignore | ||||
|             else: | ||||
|                 if not hasattr(handler, "is_websocket"): | ||||
|                     raise ServerError( | ||||
|   | ||||
| @@ -40,8 +40,6 @@ FULL_COLOR_LOGO = """ | ||||
|  | ||||
| """  # 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-?]*[ -/]*[@-~])") | ||||
|  | ||||
|  | ||||
|   | ||||
| @@ -9,7 +9,6 @@ from sanic.mixins.listeners import ListenerMixin | ||||
| 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_\-]*$") | ||||
| @@ -17,7 +16,6 @@ VALID_NAME = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_\-]*$") | ||||
|  | ||||
| class BaseSanic( | ||||
|     RouteMixin, | ||||
|     StaticMixin, | ||||
|     MiddlewareMixin, | ||||
|     ListenerMixin, | ||||
|     ExceptionMixin, | ||||
|   | ||||
| @@ -304,6 +304,9 @@ class Blueprint(BaseSanic): | ||||
|  | ||||
|         # Routes | ||||
|         for future in self._future_routes: | ||||
|             # attach the blueprint name to the handler so that it can be | ||||
|             # prefixed properly in the router | ||||
|             future.handler.__blueprintname__ = self.name | ||||
|             # Prepend the blueprint URI prefix if available | ||||
|             uri = self._setup_uri(future.uri, url_prefix) | ||||
|  | ||||
|   | ||||
| @@ -12,7 +12,6 @@ Setting ``app.config.FALLBACK_ERROR_FORMAT = "auto"`` will enable a switch that | ||||
| will attempt to provide an appropriate response format based upon the | ||||
| request type. | ||||
| """ | ||||
| from __future__ import annotations | ||||
|  | ||||
| import sys | ||||
| import typing as t | ||||
| @@ -22,7 +21,8 @@ from traceback import extract_tb | ||||
|  | ||||
| from sanic.exceptions import BadRequest, SanicException | ||||
| from sanic.helpers import STATUS_CODES | ||||
| from sanic.response import html, json, text | ||||
| from sanic.request import Request | ||||
| from sanic.response import HTTPResponse, html, json, text | ||||
|  | ||||
|  | ||||
| dumps: t.Callable[..., str] | ||||
| @@ -33,8 +33,6 @@ try: | ||||
| except ImportError:  # noqa | ||||
|     from json import dumps | ||||
|  | ||||
| if t.TYPE_CHECKING: | ||||
|     from sanic import HTTPResponse, Request | ||||
|  | ||||
| DEFAULT_FORMAT = "auto" | ||||
| FALLBACK_TEXT = ( | ||||
| @@ -406,13 +404,16 @@ CONTENT_TYPE_BY_RENDERERS = { | ||||
|     v: k for k, v in RENDERERS_BY_CONTENT_TYPE.items() | ||||
| } | ||||
|  | ||||
| # Handler source code is checked for which response types it returns with the | ||||
| # route error_format="auto" (default) to determine which format to use. | ||||
| RESPONSE_MAPPING = { | ||||
|     "empty": "html", | ||||
|     "json": "json", | ||||
|     "text": "text", | ||||
|     "raw": "text", | ||||
|     "html": "html", | ||||
|     "JSONResponse": "json", | ||||
|     "file": "html", | ||||
|     "file_stream": "text", | ||||
|     "stream": "text", | ||||
|     "redirect": "html", | ||||
|     "text/plain": "text", | ||||
|     "text/html": "html", | ||||
|     "application/json": "json", | ||||
|   | ||||
| @@ -3,6 +3,11 @@ from __future__ import annotations | ||||
| from typing import Dict, List, Optional, Tuple, Type | ||||
| 
 | ||||
| from sanic.errorpages import BaseRenderer, TextRenderer, exception_response | ||||
| from sanic.exceptions import ( | ||||
|     HeaderNotFound, | ||||
|     InvalidRangeType, | ||||
|     RangeNotSatisfiable, | ||||
| ) | ||||
| from sanic.log import deprecation, error_logger | ||||
| from sanic.models.handler_types import RouteHandler | ||||
| from sanic.response import text | ||||
| @@ -18,6 +23,7 @@ class ErrorHandler: | ||||
|     by the developers to perform a wide range of tasks from recording the error | ||||
|     stats to reporting them to an external service that can be used for | ||||
|     realtime alerting system. | ||||
| 
 | ||||
|     """ | ||||
| 
 | ||||
|     def __init__( | ||||
| @@ -190,3 +196,74 @@ class ErrorHandler: | ||||
|             error_logger.exception( | ||||
|                 "Exception occurred while handling uri: %s", url | ||||
|             ) | ||||
| 
 | ||||
| 
 | ||||
| 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 | ||||
| @@ -1,10 +0,0 @@ | ||||
| from .content_range import ContentRangeHandler | ||||
| from .directory import DirectoryHandler | ||||
| from .error import ErrorHandler | ||||
|  | ||||
|  | ||||
| __all__ = ( | ||||
|     "ContentRangeHandler", | ||||
|     "DirectoryHandler", | ||||
|     "ErrorHandler", | ||||
| ) | ||||
| @@ -1,78 +0,0 @@ | ||||
| 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 | ||||
| @@ -1,84 +0,0 @@ | ||||
| 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) | ||||
							
								
								
									
										292
									
								
								sanic/headers.py
									
									
									
									
									
								
							
							
						
						
									
										292
									
								
								sanic/headers.py
									
									
									
									
									
								
							| @@ -35,96 +35,141 @@ _host_re = re.compile( | ||||
|  | ||||
| def parse_arg_as_accept(f): | ||||
|     def func(self, other, *args, **kwargs): | ||||
|         if not isinstance(other, MediaType) and other: | ||||
|             other = MediaType._parse(other) | ||||
|         if not isinstance(other, Accept) and other: | ||||
|             other = Accept.parse(other) | ||||
|         return f(self, other, *args, **kwargs) | ||||
|  | ||||
|     return func | ||||
|  | ||||
|  | ||||
| class MediaType: | ||||
|     """A media type, as used in the Accept header.""" | ||||
| class MediaType(str): | ||||
|     def __new__(cls, value: str): | ||||
|         return str.__new__(cls, value) | ||||
|  | ||||
|     def __init__(self, value: str) -> None: | ||||
|         self.value = value | ||||
|         self.is_wildcard = self.check_if_wildcard(value) | ||||
|  | ||||
|     def __eq__(self, other): | ||||
|         if self.is_wildcard: | ||||
|             return True | ||||
|  | ||||
|         if self.match(other): | ||||
|             return True | ||||
|  | ||||
|         other_is_wildcard = ( | ||||
|             other.is_wildcard | ||||
|             if isinstance(other, MediaType) | ||||
|             else self.check_if_wildcard(other) | ||||
|         ) | ||||
|  | ||||
|         return other_is_wildcard | ||||
|  | ||||
|     def match(self, other): | ||||
|         other_value = other.value if isinstance(other, MediaType) else other | ||||
|         return self.value == other_value | ||||
|  | ||||
|     @staticmethod | ||||
|     def check_if_wildcard(value): | ||||
|         return value == "*" | ||||
|  | ||||
|  | ||||
| class Accept(str): | ||||
|     def __new__(cls, value: str, *args, **kwargs): | ||||
|         return str.__new__(cls, value) | ||||
|  | ||||
|     def __init__( | ||||
|         self, | ||||
|         type_: str, | ||||
|         subtype: str, | ||||
|         **params: str, | ||||
|         value: str, | ||||
|         type_: MediaType, | ||||
|         subtype: MediaType, | ||||
|         *, | ||||
|         q: str = "1.0", | ||||
|         **kwargs: str, | ||||
|     ): | ||||
|         qvalue = float(q) | ||||
|         if qvalue > 1 or qvalue < 0: | ||||
|             raise InvalidHeader( | ||||
|                 f"Accept header qvalue must be between 0 and 1, not: {qvalue}" | ||||
|             ) | ||||
|         self.value = value | ||||
|         self.type_ = type_ | ||||
|         self.subtype = subtype | ||||
|         self.q = float(params.get("q", "1.0")) | ||||
|         self.params = params | ||||
|         self.mime = f"{type_}/{subtype}" | ||||
|         self.qvalue = qvalue | ||||
|         self.params = kwargs | ||||
|  | ||||
|     def __repr__(self): | ||||
|         return self.mime + "".join(f";{k}={v}" for k, v in self.params.items()) | ||||
|  | ||||
|     def __eq__(self, other): | ||||
|         """Check for mime (str or MediaType) identical type/subtype.""" | ||||
|         if isinstance(other, str): | ||||
|             return self.mime == other | ||||
|         if isinstance(other, MediaType): | ||||
|             return self.mime == other.mime | ||||
|     def _compare(self, other, method): | ||||
|         try: | ||||
|             return method(self.qvalue, other.qvalue) | ||||
|         except (AttributeError, TypeError): | ||||
|             return NotImplemented | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __lt__(self, other: Union[str, Accept]): | ||||
|         return self._compare(other, lambda s, o: s < o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __le__(self, other: Union[str, Accept]): | ||||
|         return self._compare(other, lambda s, o: s <= o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __eq__(self, other: Union[str, Accept]):  # type: ignore | ||||
|         return self._compare(other, lambda s, o: s == o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __ge__(self, other: Union[str, Accept]): | ||||
|         return self._compare(other, lambda s, o: s >= o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __gt__(self, other: Union[str, Accept]): | ||||
|         return self._compare(other, lambda s, o: s > o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def __ne__(self, other: Union[str, Accept]):  # type: ignore | ||||
|         return self._compare(other, lambda s, o: s != o) | ||||
|  | ||||
|     @parse_arg_as_accept | ||||
|     def match( | ||||
|         self, | ||||
|         mime: str, | ||||
|         allow_type_wildcard=True, | ||||
|         allow_subtype_wildcard=True, | ||||
|     ) -> Optional[MediaType]: | ||||
|         """Check if this media type matches the given mime type/subtype. | ||||
|  | ||||
|         Wildcards are supported both ways on both type and subtype. | ||||
|  | ||||
|         Note:  Use the `==` operator instead to check for literal matches | ||||
|         without expanding wildcards. | ||||
|  | ||||
|         @param media_type: A type/subtype string to match. | ||||
|         @return `self` if the media types are compatible, else `None` | ||||
|         """ | ||||
|         mt = MediaType._parse(mime) | ||||
|         return ( | ||||
|             self | ||||
|             if ( | ||||
|                 # Subtype match | ||||
|                 (self.subtype in (mt.subtype, "*") or mt.subtype == "*") | ||||
|                 # Type match | ||||
|                 and (self.type_ in (mt.type_, "*") or mt.type_ == "*") | ||||
|                 # Allow disabling wildcards (backwards compatibility with tests) | ||||
|                 and ( | ||||
|                     allow_type_wildcard | ||||
|                     or self.type_ != "*" | ||||
|                     and mt.type_ != "*" | ||||
|                 ) | ||||
|                 and ( | ||||
|                     allow_subtype_wildcard | ||||
|                     or self.subtype != "*" | ||||
|                     and mt.subtype != "*" | ||||
|         other, | ||||
|         *, | ||||
|         allow_type_wildcard: bool = True, | ||||
|         allow_subtype_wildcard: bool = True, | ||||
|     ) -> bool: | ||||
|         type_match = ( | ||||
|             self.type_ == other.type_ | ||||
|             if allow_type_wildcard | ||||
|             else ( | ||||
|                 self.type_.match(other.type_) | ||||
|                 and not self.type_.is_wildcard | ||||
|                 and not other.type_.is_wildcard | ||||
|             ) | ||||
|         ) | ||||
|             else None | ||||
|         subtype_match = ( | ||||
|             self.subtype == other.subtype | ||||
|             if allow_subtype_wildcard | ||||
|             else ( | ||||
|                 self.subtype.match(other.subtype) | ||||
|                 and not self.subtype.is_wildcard | ||||
|                 and not other.subtype.is_wildcard | ||||
|             ) | ||||
|         ) | ||||
|  | ||||
|     @property | ||||
|     def has_wildcard(self) -> bool: | ||||
|         """Return True if this media type has a wildcard in it.""" | ||||
|         return "*" in (self.subtype, self.type_) | ||||
|  | ||||
|     @property | ||||
|     def is_wildcard(self) -> bool: | ||||
|         """Return True if this is the wildcard `*/*`""" | ||||
|         return self.type_ == "*" and self.subtype == "*" | ||||
|         return type_match and subtype_match | ||||
|  | ||||
|     @classmethod | ||||
|     def _parse(cls, mime_with_params: str) -> MediaType: | ||||
|         mtype = mime_with_params.strip() | ||||
|     def parse(cls, raw: str) -> Accept: | ||||
|         invalid = False | ||||
|         mtype = raw.strip() | ||||
|  | ||||
|         try: | ||||
|             media, *raw_params = mtype.split(";") | ||||
|         type_, subtype = media.split("/", 1) | ||||
|         if not type_ or not subtype: | ||||
|             raise ValueError(f"Invalid media type: {mtype}") | ||||
|             type_, subtype = media.split("/") | ||||
|         except ValueError: | ||||
|             invalid = True | ||||
|  | ||||
|         if invalid or not type_ or not subtype: | ||||
|             raise InvalidHeader(f"Header contains invalid Accept value: {raw}") | ||||
|  | ||||
|         params = dict( | ||||
|             [ | ||||
| @@ -133,79 +178,28 @@ class MediaType: | ||||
|             ] | ||||
|         ) | ||||
|  | ||||
|         return cls(type_.lstrip(), subtype.rstrip(), **params) | ||||
|         return cls(mtype, MediaType(type_), MediaType(subtype), **params) | ||||
|  | ||||
|  | ||||
| class Matched(str): | ||||
|     """A matching result of a MIME string against a MediaType.""" | ||||
| class AcceptContainer(list): | ||||
|     def __contains__(self, o: object) -> bool: | ||||
|         return any(item.match(o) for item in self) | ||||
|  | ||||
|     def __new__(cls, mime: str, m: Optional[MediaType]): | ||||
|         return super().__new__(cls, mime) | ||||
|  | ||||
|     def __init__(self, mime: str, m: Optional[MediaType]): | ||||
|         self.m = m | ||||
|  | ||||
|     def __repr__(self): | ||||
|         return f"<{self} matched {self.m}>" if self else "<no match>" | ||||
|  | ||||
|  | ||||
| class AcceptList(list): | ||||
|     """A list of media types, as used in the Accept header. | ||||
|  | ||||
|     The Accept header entries are listed in order of preference, starting | ||||
|     with the most preferred. This class is a list of `MediaType` objects, | ||||
|     that encapsulate also the q value or any other parameters. | ||||
|  | ||||
|     Two separate methods are provided for searching the list: | ||||
|     - 'match' for finding the most preferred match (wildcards supported) | ||||
|     -  operator 'in' for checking explicit matches (wildcards as literals) | ||||
|     """ | ||||
|  | ||||
|     def match(self, *mimes: str) -> Matched: | ||||
|         """Find a media type accepted by the client. | ||||
|  | ||||
|         This method can be used to find which of the media types requested by | ||||
|         the client is most preferred against the ones given as arguments. | ||||
|  | ||||
|         The ordering of preference is set by: | ||||
|         1. The q values on the Accept header, and those being equal, | ||||
|         2. The order of the arguments (first is most preferred), and | ||||
|         3. The first matching entry on the Accept header. | ||||
|  | ||||
|         Wildcards are matched both ways. A match is usually found, as the | ||||
|         Accept headers typically include `*/*`, in particular if the header | ||||
|         is missing, is not manually set, or if the client is a browser. | ||||
|  | ||||
|         Note: the returned object behaves as a string of the mime argument | ||||
|         that matched, and is empty/falsy if no match was found. The matched | ||||
|         header entry `MediaType` or `None` is available as the `m` attribute. | ||||
|  | ||||
|         @param mimes: Any MIME types to search for in order of preference. | ||||
|         @return A match object with the mime string and the MediaType object. | ||||
|         """ | ||||
|         l = sorted( | ||||
|             [ | ||||
|                 (-acc.q, i, j, mime, acc)  # Sort by -q, i, j | ||||
|                 for j, acc in enumerate(self) | ||||
|                 for i, mime in enumerate(mimes) | ||||
|                 if acc.match(mime) | ||||
|             ] | ||||
|     def match( | ||||
|         self, | ||||
|         o: object, | ||||
|         *, | ||||
|         allow_type_wildcard: bool = True, | ||||
|         allow_subtype_wildcard: bool = True, | ||||
|     ) -> bool: | ||||
|         return any( | ||||
|             item.match( | ||||
|                 o, | ||||
|                 allow_type_wildcard=allow_type_wildcard, | ||||
|                 allow_subtype_wildcard=allow_subtype_wildcard, | ||||
|             ) | ||||
|             for item in self | ||||
|         ) | ||||
|         return Matched(*(l[0][3:] if l else ("", None))) | ||||
|  | ||||
|  | ||||
| def parse_accept(accept: str) -> AcceptList: | ||||
|     """Parse an Accept header and order the acceptable media types in | ||||
|     accorsing to RFC 7231, s. 5.3.2 | ||||
|     https://datatracker.ietf.org/doc/html/rfc7231#section-5.3.2 | ||||
|     """ | ||||
|     if not accept: | ||||
|         return AcceptList() | ||||
|     try: | ||||
|         a = [MediaType._parse(mtype) for mtype in accept.split(",")] | ||||
|         return AcceptList(sorted(a, key=lambda mtype: -mtype.q)) | ||||
|     except ValueError: | ||||
|         raise InvalidHeader(f"Invalid header value in Accept: {accept}") | ||||
|  | ||||
|  | ||||
| def parse_content_header(value: str) -> Tuple[str, Options]: | ||||
| @@ -374,6 +368,34 @@ def format_http1_response(status: int, headers: HeaderBytesIterable) -> bytes: | ||||
|     return ret | ||||
|  | ||||
|  | ||||
| def _sort_accept_value(accept: Accept): | ||||
|     return ( | ||||
|         accept.qvalue, | ||||
|         len(accept.params), | ||||
|         accept.subtype != "*", | ||||
|         accept.type_ != "*", | ||||
|     ) | ||||
|  | ||||
|  | ||||
| def parse_accept(accept: str) -> AcceptContainer: | ||||
|     """Parse an Accept header and order the acceptable media types in | ||||
|     accorsing to RFC 7231, s. 5.3.2 | ||||
|     https://datatracker.ietf.org/doc/html/rfc7231#section-5.3.2 | ||||
|     """ | ||||
|     media_types = accept.split(",") | ||||
|     accept_list: List[Accept] = [] | ||||
|  | ||||
|     for mtype in media_types: | ||||
|         if not mtype: | ||||
|             continue | ||||
|  | ||||
|         accept_list.append(Accept.parse(mtype)) | ||||
|  | ||||
|     return AcceptContainer( | ||||
|         sorted(accept_list, key=_sort_accept_value, reverse=True) | ||||
|     ) | ||||
|  | ||||
|  | ||||
| def parse_credentials( | ||||
|     header: Optional[str], | ||||
|     prefixes: Union[List, Tuple, Set] = None, | ||||
|   | ||||
| @@ -1,35 +0,0 @@ | ||||
| from typing import Optional | ||||
|  | ||||
| from sanic.base.meta import SanicMeta | ||||
|  | ||||
|  | ||||
| class BaseMixin(metaclass=SanicMeta): | ||||
|     name: str | ||||
|     strict_slashes: Optional[bool] | ||||
|  | ||||
|     def _generate_name(self, *objects) -> str: | ||||
|         name = None | ||||
|  | ||||
|         for obj in objects: | ||||
|             if obj: | ||||
|                 if isinstance(obj, str): | ||||
|                     name = obj | ||||
|                     break | ||||
|  | ||||
|                 try: | ||||
|                     name = obj.name | ||||
|                 except AttributeError: | ||||
|                     try: | ||||
|                         name = obj.__name__ | ||||
|                     except AttributeError: | ||||
|                         continue | ||||
|                 else: | ||||
|                     break | ||||
|  | ||||
|         if not name:  # noqa | ||||
|             raise ValueError("Could not generate a name for handler") | ||||
|  | ||||
|         if not name.startswith(f"{self.name}."): | ||||
|             name = f"{self.name}.{name}" | ||||
|  | ||||
|         return name | ||||
| @@ -1,6 +1,11 @@ | ||||
| from ast import NodeVisitor, Return, parse | ||||
| from contextlib import suppress | ||||
| from email.utils import formatdate | ||||
| from functools import partial, wraps | ||||
| from inspect import getsource, signature | ||||
| from mimetypes import guess_type | ||||
| from os import path | ||||
| from pathlib import Path, PurePath | ||||
| from textwrap import dedent | ||||
| from typing import ( | ||||
|     Any, | ||||
| @@ -14,15 +19,20 @@ from typing import ( | ||||
|     Union, | ||||
|     cast, | ||||
| ) | ||||
| from urllib.parse import unquote | ||||
|  | ||||
| from sanic_routing.route import Route | ||||
|  | ||||
| from sanic.base.meta import SanicMeta | ||||
| from sanic.constants import HTTP_METHODS | ||||
| from sanic.compat import stat_async | ||||
| from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE, HTTP_METHODS | ||||
| from sanic.errorpages import RESPONSE_MAPPING | ||||
| from sanic.mixins.base import BaseMixin | ||||
| from sanic.exceptions import FileNotFound, HeaderNotFound, RangeNotSatisfiable | ||||
| from sanic.handlers import ContentRangeHandler | ||||
| from sanic.log import error_logger | ||||
| from sanic.models.futures import FutureRoute, FutureStatic | ||||
| from sanic.models.handler_types import RouteHandler | ||||
| from sanic.response import HTTPResponse, file, file_stream, validate_file | ||||
| from sanic.types import HashableDict | ||||
|  | ||||
|  | ||||
| @@ -31,14 +41,20 @@ RouteWrapper = Callable[ | ||||
| ] | ||||
|  | ||||
|  | ||||
| class RouteMixin(BaseMixin, metaclass=SanicMeta): | ||||
| class RouteMixin(metaclass=SanicMeta): | ||||
|     name: str | ||||
|  | ||||
|     def __init__(self, *args, **kwargs) -> None: | ||||
|         self._future_routes: Set[FutureRoute] = set() | ||||
|         self._future_statics: Set[FutureStatic] = set() | ||||
|         self.strict_slashes: Optional[bool] = False | ||||
|  | ||||
|     def _apply_route(self, route: FutureRoute) -> List[Route]: | ||||
|         raise NotImplementedError  # noqa | ||||
|  | ||||
|     def _apply_static(self, static: FutureStatic) -> Route: | ||||
|         raise NotImplementedError  # noqa | ||||
|  | ||||
|     def route( | ||||
|         self, | ||||
|         uri: str, | ||||
| @@ -672,6 +688,324 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta): | ||||
|             **ctx_kwargs, | ||||
|         )(handler) | ||||
|  | ||||
|     def static( | ||||
|         self, | ||||
|         uri: str, | ||||
|         file_or_directory: Union[str, bytes, PurePath], | ||||
|         pattern: str = r"/?.+", | ||||
|         use_modified_since: bool = True, | ||||
|         use_content_range: bool = False, | ||||
|         stream_large_files: bool = False, | ||||
|         name: str = "static", | ||||
|         host: Optional[str] = None, | ||||
|         strict_slashes: Optional[bool] = None, | ||||
|         content_type: Optional[bool] = None, | ||||
|         apply: bool = True, | ||||
|         resource_type: Optional[str] = None, | ||||
|     ): | ||||
|         """ | ||||
|         Register a root to serve files from. The input can either be a | ||||
|         file or a directory. This method will enable an easy and simple way | ||||
|         to setup the :class:`Route` necessary to serve the static files. | ||||
|  | ||||
|         :param uri: URL path to be used for serving static content | ||||
|         :param file_or_directory: Path for the Static file/directory with | ||||
|             static files | ||||
|         :param pattern: Regex Pattern identifying the valid static files | ||||
|         :param use_modified_since: If true, send file modified time, and return | ||||
|             not modified if the browser's matches the server's | ||||
|         :param use_content_range: If true, process header for range requests | ||||
|             and sends the file part that is requested | ||||
|         :param stream_large_files: If true, use the | ||||
|             :func:`StreamingHTTPResponse.file_stream` handler rather | ||||
|             than the :func:`HTTPResponse.file` handler to send the file. | ||||
|             If this is an integer, this represents the threshold size to | ||||
|             switch to :func:`StreamingHTTPResponse.file_stream` | ||||
|         :param name: user defined name used for url_for | ||||
|         :param host: Host IP or FQDN for the service to use | ||||
|         :param strict_slashes: Instruct :class:`Sanic` to check if the request | ||||
|             URLs need to terminate with a */* | ||||
|         :param content_type: user defined content type for header | ||||
|         :return: routes registered on the router | ||||
|         :rtype: List[sanic.router.Route] | ||||
|         """ | ||||
|  | ||||
|         name = self._generate_name(name) | ||||
|  | ||||
|         if strict_slashes is None and self.strict_slashes is not None: | ||||
|             strict_slashes = self.strict_slashes | ||||
|  | ||||
|         if not isinstance(file_or_directory, (str, bytes, PurePath)): | ||||
|             raise ValueError( | ||||
|                 f"Static route must be a valid path, not {file_or_directory}" | ||||
|             ) | ||||
|  | ||||
|         static = FutureStatic( | ||||
|             uri, | ||||
|             file_or_directory, | ||||
|             pattern, | ||||
|             use_modified_since, | ||||
|             use_content_range, | ||||
|             stream_large_files, | ||||
|             name, | ||||
|             host, | ||||
|             strict_slashes, | ||||
|             content_type, | ||||
|             resource_type, | ||||
|         ) | ||||
|         self._future_statics.add(static) | ||||
|  | ||||
|         if apply: | ||||
|             self._apply_static(static) | ||||
|  | ||||
|     def _generate_name(self, *objects) -> str: | ||||
|         name = None | ||||
|  | ||||
|         for obj in objects: | ||||
|             if obj: | ||||
|                 if isinstance(obj, str): | ||||
|                     name = obj | ||||
|                     break | ||||
|  | ||||
|                 try: | ||||
|                     name = obj.name | ||||
|                 except AttributeError: | ||||
|                     try: | ||||
|                         name = obj.__name__ | ||||
|                     except AttributeError: | ||||
|                         continue | ||||
|                 else: | ||||
|                     break | ||||
|  | ||||
|         if not name:  # noqa | ||||
|             raise ValueError("Could not generate a name for handler") | ||||
|  | ||||
|         if not name.startswith(f"{self.name}."): | ||||
|             name = f"{self.name}.{name}" | ||||
|  | ||||
|         return name | ||||
|  | ||||
|     async def _get_file_path(self, file_or_directory, __file_uri__, not_found): | ||||
|         file_path_raw = Path(unquote(file_or_directory)) | ||||
|         root_path = file_path = file_path_raw.resolve() | ||||
|  | ||||
|         if __file_uri__: | ||||
|             # Strip all / that in the beginning of the URL to help prevent | ||||
|             # python from herping a derp and treating the uri as an | ||||
|             # absolute path | ||||
|             unquoted_file_uri = unquote(__file_uri__).lstrip("/") | ||||
|             file_path_raw = Path(file_or_directory, unquoted_file_uri) | ||||
|             file_path = file_path_raw.resolve() | ||||
|             if ( | ||||
|                 file_path < root_path and not file_path_raw.is_symlink() | ||||
|             ) or ".." in file_path_raw.parts: | ||||
|                 error_logger.exception( | ||||
|                     f"File not found: path={file_or_directory}, " | ||||
|                     f"relative_url={__file_uri__}" | ||||
|                 ) | ||||
|                 raise not_found | ||||
|  | ||||
|         try: | ||||
|             file_path.relative_to(root_path) | ||||
|         except ValueError: | ||||
|             if not file_path_raw.is_symlink(): | ||||
|                 error_logger.exception( | ||||
|                     f"File not found: path={file_or_directory}, " | ||||
|                     f"relative_url={__file_uri__}" | ||||
|                 ) | ||||
|                 raise not_found | ||||
|         return file_path | ||||
|  | ||||
|     async def _static_request_handler( | ||||
|         self, | ||||
|         file_or_directory, | ||||
|         use_modified_since, | ||||
|         use_content_range, | ||||
|         stream_large_files, | ||||
|         request, | ||||
|         content_type=None, | ||||
|         __file_uri__=None, | ||||
|     ): | ||||
|         not_found = FileNotFound( | ||||
|             "File not found", | ||||
|             path=file_or_directory, | ||||
|             relative_url=__file_uri__, | ||||
|         ) | ||||
|  | ||||
|         # Merge served directory and requested file if provided | ||||
|         file_path = await self._get_file_path( | ||||
|             file_or_directory, __file_uri__, not_found | ||||
|         ) | ||||
|  | ||||
|         try: | ||||
|             headers = {} | ||||
|             # Check if the client has been sent this file before | ||||
|             # and it has not been modified since | ||||
|             stats = None | ||||
|             if use_modified_since: | ||||
|                 stats = await stat_async(file_path) | ||||
|                 modified_since = stats.st_mtime | ||||
|                 response = await validate_file(request.headers, modified_since) | ||||
|                 if response: | ||||
|                     return response | ||||
|                 headers["Last-Modified"] = formatdate( | ||||
|                     modified_since, usegmt=True | ||||
|                 ) | ||||
|             _range = None | ||||
|             if use_content_range: | ||||
|                 _range = None | ||||
|                 if not stats: | ||||
|                     stats = await stat_async(file_path) | ||||
|                 headers["Accept-Ranges"] = "bytes" | ||||
|                 headers["Content-Length"] = str(stats.st_size) | ||||
|                 if request.method != "HEAD": | ||||
|                     try: | ||||
|                         _range = ContentRangeHandler(request, stats) | ||||
|                     except HeaderNotFound: | ||||
|                         pass | ||||
|                     else: | ||||
|                         del headers["Content-Length"] | ||||
|                         headers.update(_range.headers) | ||||
|  | ||||
|             if "content-type" not in headers: | ||||
|                 content_type = ( | ||||
|                     content_type | ||||
|                     or guess_type(file_path)[0] | ||||
|                     or DEFAULT_HTTP_CONTENT_TYPE | ||||
|                 ) | ||||
|  | ||||
|                 if "charset=" not in content_type and ( | ||||
|                     content_type.startswith("text/") | ||||
|                     or content_type == "application/javascript" | ||||
|                 ): | ||||
|                     content_type += "; charset=utf-8" | ||||
|  | ||||
|                 headers["Content-Type"] = content_type | ||||
|  | ||||
|             if request.method == "HEAD": | ||||
|                 return HTTPResponse(headers=headers) | ||||
|             else: | ||||
|                 if stream_large_files: | ||||
|                     if type(stream_large_files) == int: | ||||
|                         threshold = stream_large_files | ||||
|                     else: | ||||
|                         threshold = 1024 * 1024 | ||||
|  | ||||
|                     if not stats: | ||||
|                         stats = await stat_async(file_path) | ||||
|                     if stats.st_size >= threshold: | ||||
|                         return await file_stream( | ||||
|                             file_path, headers=headers, _range=_range | ||||
|                         ) | ||||
|                 return await file(file_path, headers=headers, _range=_range) | ||||
|         except RangeNotSatisfiable: | ||||
|             raise | ||||
|         except FileNotFoundError: | ||||
|             raise not_found | ||||
|         except Exception: | ||||
|             error_logger.exception( | ||||
|                 f"Exception in static request handler: " | ||||
|                 f"path={file_or_directory}, " | ||||
|                 f"relative_url={__file_uri__}" | ||||
|             ) | ||||
|             raise | ||||
|  | ||||
|     def _register_static( | ||||
|         self, | ||||
|         static: FutureStatic, | ||||
|     ): | ||||
|         # TODO: Though sanic is not a file server, I feel like we should | ||||
|         # at least make a good effort here.  Modified-since is nice, but | ||||
|         # we could also look into etags, expires, and caching | ||||
|         """ | ||||
|         Register a static directory handler with Sanic by adding a route to the | ||||
|         router and registering a handler. | ||||
|  | ||||
|         :param app: Sanic | ||||
|         :param file_or_directory: File or directory path to serve from | ||||
|         :type file_or_directory: Union[str,bytes,Path] | ||||
|         :param uri: URL to serve from | ||||
|         :type uri: str | ||||
|         :param pattern: regular expression used to match files in the URL | ||||
|         :param use_modified_since: If true, send file modified time, and return | ||||
|                                 not modified if the browser's matches the | ||||
|                                 server's | ||||
|         :param use_content_range: If true, process header for range requests | ||||
|                                 and sends the file part that is requested | ||||
|         :param stream_large_files: If true, use the file_stream() handler | ||||
|                                 rather than the file() handler to send the file | ||||
|                                 If this is an integer, this represents the | ||||
|                                 threshold size to switch to file_stream() | ||||
|         :param name: user defined name used for url_for | ||||
|         :type name: str | ||||
|         :param content_type: user defined content type for header | ||||
|         :return: registered static routes | ||||
|         :rtype: List[sanic.router.Route] | ||||
|         """ | ||||
|  | ||||
|         if isinstance(static.file_or_directory, bytes): | ||||
|             file_or_directory = static.file_or_directory.decode("utf-8") | ||||
|         elif isinstance(static.file_or_directory, PurePath): | ||||
|             file_or_directory = str(static.file_or_directory) | ||||
|         elif not isinstance(static.file_or_directory, str): | ||||
|             raise ValueError("Invalid file path string.") | ||||
|         else: | ||||
|             file_or_directory = static.file_or_directory | ||||
|  | ||||
|         uri = static.uri | ||||
|         name = static.name | ||||
|         # If we're not trying to match a file directly, | ||||
|         # serve from the folder | ||||
|         if not static.resource_type: | ||||
|             if not path.isfile(file_or_directory): | ||||
|                 uri = uri.rstrip("/") | ||||
|                 uri += "/<__file_uri__:path>" | ||||
|         elif static.resource_type == "dir": | ||||
|             if path.isfile(file_or_directory): | ||||
|                 raise TypeError( | ||||
|                     "Resource type improperly identified as directory. " | ||||
|                     f"'{file_or_directory}'" | ||||
|                 ) | ||||
|             uri = uri.rstrip("/") | ||||
|             uri += "/<__file_uri__:path>" | ||||
|         elif static.resource_type == "file" and not path.isfile( | ||||
|             file_or_directory | ||||
|         ): | ||||
|             raise TypeError( | ||||
|                 "Resource type improperly identified as file. " | ||||
|                 f"'{file_or_directory}'" | ||||
|             ) | ||||
|         elif static.resource_type != "file": | ||||
|             raise ValueError( | ||||
|                 "The resource_type should be set to 'file' or 'dir'" | ||||
|             ) | ||||
|  | ||||
|         # special prefix for static files | ||||
|         # if not static.name.startswith("_static_"): | ||||
|         #     name = f"_static_{static.name}" | ||||
|  | ||||
|         _handler = wraps(self._static_request_handler)( | ||||
|             partial( | ||||
|                 self._static_request_handler, | ||||
|                 file_or_directory, | ||||
|                 static.use_modified_since, | ||||
|                 static.use_content_range, | ||||
|                 static.stream_large_files, | ||||
|                 content_type=static.content_type, | ||||
|             ) | ||||
|         ) | ||||
|  | ||||
|         route, _ = self.route(  # type: ignore | ||||
|             uri=uri, | ||||
|             methods=["GET", "HEAD"], | ||||
|             name=name, | ||||
|             host=static.host, | ||||
|             strict_slashes=static.strict_slashes, | ||||
|             static=True, | ||||
|         )(_handler) | ||||
|  | ||||
|         return route | ||||
|  | ||||
|     def _determine_error_format(self, handler) -> str: | ||||
|         with suppress(OSError, TypeError): | ||||
|             src = dedent(getsource(handler)) | ||||
|   | ||||
| @@ -1,348 +0,0 @@ | ||||
| from email.utils import formatdate | ||||
| from functools import partial, wraps | ||||
| from mimetypes import guess_type | ||||
| from os import PathLike, path | ||||
| from pathlib import Path, PurePath | ||||
| from typing import Optional, Sequence, Set, Union, cast | ||||
| from urllib.parse import unquote | ||||
|  | ||||
| from sanic_routing.route import Route | ||||
|  | ||||
| from sanic.base.meta import SanicMeta | ||||
| from sanic.compat import stat_async | ||||
| from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE | ||||
| from sanic.exceptions import FileNotFound, HeaderNotFound, RangeNotSatisfiable | ||||
| from sanic.handlers import ContentRangeHandler | ||||
| from sanic.handlers.directory import DirectoryHandler | ||||
| from sanic.log import deprecation, error_logger | ||||
| from sanic.mixins.base import BaseMixin | ||||
| from sanic.models.futures import FutureStatic | ||||
| from sanic.request import Request | ||||
| from sanic.response import HTTPResponse, file, file_stream, validate_file | ||||
|  | ||||
|  | ||||
| class StaticMixin(BaseMixin, metaclass=SanicMeta): | ||||
|     def __init__(self, *args, **kwargs) -> None: | ||||
|         self._future_statics: Set[FutureStatic] = set() | ||||
|  | ||||
|     def _apply_static(self, static: FutureStatic) -> Route: | ||||
|         raise NotImplementedError  # noqa | ||||
|  | ||||
|     def static( | ||||
|         self, | ||||
|         uri: str, | ||||
|         file_or_directory: Union[PathLike, str, bytes], | ||||
|         pattern: str = r"/?.+", | ||||
|         use_modified_since: bool = True, | ||||
|         use_content_range: bool = False, | ||||
|         stream_large_files: Union[bool, int] = False, | ||||
|         name: str = "static", | ||||
|         host: Optional[str] = None, | ||||
|         strict_slashes: Optional[bool] = None, | ||||
|         content_type: Optional[str] = None, | ||||
|         apply: bool = True, | ||||
|         resource_type: Optional[str] = None, | ||||
|         index: Optional[Union[str, Sequence[str]]] = None, | ||||
|         directory_view: bool = False, | ||||
|         directory_handler: Optional[DirectoryHandler] = None, | ||||
|     ): | ||||
|         """ | ||||
|         Register a root to serve files from. The input can either be a | ||||
|         file or a directory. This method will enable an easy and simple way | ||||
|         to setup the :class:`Route` necessary to serve the static files. | ||||
|  | ||||
|         :param uri: URL path to be used for serving static content | ||||
|         :param file_or_directory: Path for the Static file/directory with | ||||
|             static files | ||||
|         :param pattern: Regex Pattern identifying the valid static files | ||||
|         :param use_modified_since: If true, send file modified time, and return | ||||
|             not modified if the browser's matches the server's | ||||
|         :param use_content_range: If true, process header for range requests | ||||
|             and sends the file part that is requested | ||||
|         :param stream_large_files: If true, use the | ||||
|             :func:`StreamingHTTPResponse.file_stream` handler rather | ||||
|             than the :func:`HTTPResponse.file` handler to send the file. | ||||
|             If this is an integer, this represents the threshold size to | ||||
|             switch to :func:`StreamingHTTPResponse.file_stream` | ||||
|         :param name: user defined name used for url_for | ||||
|         :param host: Host IP or FQDN for the service to use | ||||
|         :param strict_slashes: Instruct :class:`Sanic` to check if the request | ||||
|             URLs need to terminate with a */* | ||||
|         :param content_type: user defined content type for header | ||||
|         :param apply: If true, will register the route immediately | ||||
|         :param resource_type: Explicitly declare a resource to be a " | ||||
|             file" or a "dir" | ||||
|         :param index: When exposing against a directory, index is the name that | ||||
|             will be served as the default file. When multiple files names are | ||||
|             passed, then they will be tried in order. | ||||
|         :param directory_view: Whether to fallback to showing the directory | ||||
|             viewer when exposing a directory | ||||
|         :param directory_handler: An instance of :class:`DirectoryHandler` | ||||
|             that can be used for explicitly controlling and subclassing the | ||||
|             behavior of the default directory handler | ||||
|         :return: routes registered on the router | ||||
|         :rtype: List[sanic.router.Route] | ||||
|         """ | ||||
|  | ||||
|         name = self._generate_name(name) | ||||
|  | ||||
|         if strict_slashes is None and self.strict_slashes is not None: | ||||
|             strict_slashes = self.strict_slashes | ||||
|  | ||||
|         if not isinstance(file_or_directory, (str, bytes, PurePath)): | ||||
|             raise ValueError( | ||||
|                 f"Static route must be a valid path, not {file_or_directory}" | ||||
|             ) | ||||
|  | ||||
|         if isinstance(file_or_directory, bytes): | ||||
|             deprecation( | ||||
|                 "Serving a static directory with a bytes string is " | ||||
|                 "deprecated and will be removed in v22.9.", | ||||
|                 22.9, | ||||
|             ) | ||||
|             file_or_directory = cast(str, file_or_directory.decode()) | ||||
|         file_or_directory = Path(file_or_directory) | ||||
|  | ||||
|         if directory_handler and (directory_view or index): | ||||
|             raise ValueError( | ||||
|                 "When explicitly setting directory_handler, you cannot " | ||||
|                 "set either directory_view or index. Instead, pass " | ||||
|                 "these arguments to your DirectoryHandler instance." | ||||
|             ) | ||||
|  | ||||
|         if not directory_handler: | ||||
|             directory_handler = DirectoryHandler( | ||||
|                 uri=uri, | ||||
|                 directory=file_or_directory, | ||||
|                 directory_view=directory_view, | ||||
|                 index=index, | ||||
|             ) | ||||
|  | ||||
|         static = FutureStatic( | ||||
|             uri, | ||||
|             file_or_directory, | ||||
|             pattern, | ||||
|             use_modified_since, | ||||
|             use_content_range, | ||||
|             stream_large_files, | ||||
|             name, | ||||
|             host, | ||||
|             strict_slashes, | ||||
|             content_type, | ||||
|             resource_type, | ||||
|             directory_handler, | ||||
|         ) | ||||
|         self._future_statics.add(static) | ||||
|  | ||||
|         if apply: | ||||
|             self._apply_static(static) | ||||
|  | ||||
|  | ||||
| class StaticHandleMixin(metaclass=SanicMeta): | ||||
|     def _apply_static(self, static: FutureStatic) -> Route: | ||||
|         return self._register_static(static) | ||||
|  | ||||
|     def _register_static( | ||||
|         self, | ||||
|         static: FutureStatic, | ||||
|     ): | ||||
|         # TODO: Though sanic is not a file server, I feel like we should | ||||
|         # at least make a good effort here.  Modified-since is nice, but | ||||
|         # we could also look into etags, expires, and caching | ||||
|         """ | ||||
|         Register a static directory handler with Sanic by adding a route to the | ||||
|         router and registering a handler. | ||||
|         """ | ||||
|  | ||||
|         if isinstance(static.file_or_directory, bytes): | ||||
|             file_or_directory = static.file_or_directory.decode("utf-8") | ||||
|         elif isinstance(static.file_or_directory, PurePath): | ||||
|             file_or_directory = str(static.file_or_directory) | ||||
|         elif not isinstance(static.file_or_directory, str): | ||||
|             raise ValueError("Invalid file path string.") | ||||
|         else: | ||||
|             file_or_directory = static.file_or_directory | ||||
|  | ||||
|         uri = static.uri | ||||
|         name = static.name | ||||
|         # If we're not trying to match a file directly, | ||||
|         # serve from the folder | ||||
|         if not static.resource_type: | ||||
|             if not path.isfile(file_or_directory): | ||||
|                 uri = uri.rstrip("/") | ||||
|                 uri += "/<__file_uri__:path>" | ||||
|         elif static.resource_type == "dir": | ||||
|             if path.isfile(file_or_directory): | ||||
|                 raise TypeError( | ||||
|                     "Resource type improperly identified as directory. " | ||||
|                     f"'{file_or_directory}'" | ||||
|                 ) | ||||
|             uri = uri.rstrip("/") | ||||
|             uri += "/<__file_uri__:path>" | ||||
|         elif static.resource_type == "file" and not path.isfile( | ||||
|             file_or_directory | ||||
|         ): | ||||
|             raise TypeError( | ||||
|                 "Resource type improperly identified as file. " | ||||
|                 f"'{file_or_directory}'" | ||||
|             ) | ||||
|         elif static.resource_type != "file": | ||||
|             raise ValueError( | ||||
|                 "The resource_type should be set to 'file' or 'dir'" | ||||
|             ) | ||||
|  | ||||
|         # special prefix for static files | ||||
|         # if not static.name.startswith("_static_"): | ||||
|         #     name = f"_static_{static.name}" | ||||
|  | ||||
|         _handler = wraps(self._static_request_handler)( | ||||
|             partial( | ||||
|                 self._static_request_handler, | ||||
|                 file_or_directory=file_or_directory, | ||||
|                 use_modified_since=static.use_modified_since, | ||||
|                 use_content_range=static.use_content_range, | ||||
|                 stream_large_files=static.stream_large_files, | ||||
|                 content_type=static.content_type, | ||||
|                 directory_handler=static.directory_handler, | ||||
|             ) | ||||
|         ) | ||||
|  | ||||
|         route, _ = self.route(  # type: ignore | ||||
|             uri=uri, | ||||
|             methods=["GET", "HEAD"], | ||||
|             name=name, | ||||
|             host=static.host, | ||||
|             strict_slashes=static.strict_slashes, | ||||
|             static=True, | ||||
|         )(_handler) | ||||
|  | ||||
|         return route | ||||
|  | ||||
|     async def _static_request_handler( | ||||
|         self, | ||||
|         request: Request, | ||||
|         *, | ||||
|         file_or_directory: PathLike, | ||||
|         use_modified_since: bool, | ||||
|         use_content_range: bool, | ||||
|         stream_large_files: Union[bool, int], | ||||
|         directory_handler: DirectoryHandler, | ||||
|         content_type: Optional[str] = None, | ||||
|         __file_uri__: Optional[str] = None, | ||||
|     ): | ||||
|         not_found = FileNotFound( | ||||
|             "File not found", | ||||
|             path=file_or_directory, | ||||
|             relative_url=__file_uri__, | ||||
|         ) | ||||
|  | ||||
|         # Merge served directory and requested file if provided | ||||
|         file_path = await self._get_file_path( | ||||
|             file_or_directory, __file_uri__, not_found | ||||
|         ) | ||||
|  | ||||
|         try: | ||||
|             headers = {} | ||||
|             # Check if the client has been sent this file before | ||||
|             # and it has not been modified since | ||||
|             stats = None | ||||
|             if use_modified_since: | ||||
|                 stats = await stat_async(file_path) | ||||
|                 modified_since = stats.st_mtime | ||||
|                 response = await validate_file(request.headers, modified_since) | ||||
|                 if response: | ||||
|                     return response | ||||
|                 headers["Last-Modified"] = formatdate( | ||||
|                     modified_since, usegmt=True | ||||
|                 ) | ||||
|             _range = None | ||||
|             if use_content_range: | ||||
|                 _range = None | ||||
|                 if not stats: | ||||
|                     stats = await stat_async(file_path) | ||||
|                 headers["Accept-Ranges"] = "bytes" | ||||
|                 headers["Content-Length"] = str(stats.st_size) | ||||
|                 if request.method != "HEAD": | ||||
|                     try: | ||||
|                         _range = ContentRangeHandler(request, stats) | ||||
|                     except HeaderNotFound: | ||||
|                         pass | ||||
|                     else: | ||||
|                         del headers["Content-Length"] | ||||
|                         headers.update(_range.headers) | ||||
|  | ||||
|             if "content-type" not in headers: | ||||
|                 content_type = ( | ||||
|                     content_type | ||||
|                     or guess_type(file_path)[0] | ||||
|                     or DEFAULT_HTTP_CONTENT_TYPE | ||||
|                 ) | ||||
|  | ||||
|                 if "charset=" not in content_type and ( | ||||
|                     content_type.startswith("text/") | ||||
|                     or content_type == "application/javascript" | ||||
|                 ): | ||||
|                     content_type += "; charset=utf-8" | ||||
|  | ||||
|                 headers["Content-Type"] = content_type | ||||
|  | ||||
|             if request.method == "HEAD": | ||||
|                 return HTTPResponse(headers=headers) | ||||
|             else: | ||||
|                 if stream_large_files: | ||||
|                     if isinstance(stream_large_files, bool): | ||||
|                         threshold = 1024 * 1024 | ||||
|                     else: | ||||
|                         threshold = stream_large_files | ||||
|  | ||||
|                     if not stats: | ||||
|                         stats = await stat_async(file_path) | ||||
|                     if stats.st_size >= threshold: | ||||
|                         return await file_stream( | ||||
|                             file_path, headers=headers, _range=_range | ||||
|                         ) | ||||
|                 return await file(file_path, headers=headers, _range=_range) | ||||
|         except (IsADirectoryError, PermissionError): | ||||
|             return await directory_handler.handle(request, request.path) | ||||
|         except RangeNotSatisfiable: | ||||
|             raise | ||||
|         except FileNotFoundError: | ||||
|             raise not_found | ||||
|         except Exception: | ||||
|             error_logger.exception( | ||||
|                 "Exception in static request handler: " | ||||
|                 f"path={file_or_directory}, " | ||||
|                 f"relative_url={__file_uri__}" | ||||
|             ) | ||||
|             raise | ||||
|  | ||||
|     async def _get_file_path(self, file_or_directory, __file_uri__, not_found): | ||||
|         file_path_raw = Path(unquote(file_or_directory)) | ||||
|         root_path = file_path = file_path_raw.resolve() | ||||
|  | ||||
|         if __file_uri__: | ||||
|             # Strip all / that in the beginning of the URL to help prevent | ||||
|             # python from herping a derp and treating the uri as an | ||||
|             # absolute path | ||||
|             unquoted_file_uri = unquote(__file_uri__).lstrip("/") | ||||
|             file_path_raw = Path(file_or_directory, unquoted_file_uri) | ||||
|             file_path = file_path_raw.resolve() | ||||
|             if ( | ||||
|                 file_path < root_path and not file_path_raw.is_symlink() | ||||
|             ) or ".." in file_path_raw.parts: | ||||
|                 error_logger.exception( | ||||
|                     f"File not found: path={file_or_directory}, " | ||||
|                     f"relative_url={__file_uri__}" | ||||
|                 ) | ||||
|                 raise not_found | ||||
|  | ||||
|         try: | ||||
|             file_path.relative_to(root_path) | ||||
|         except ValueError: | ||||
|             if not file_path_raw.is_symlink(): | ||||
|                 error_logger.exception( | ||||
|                     f"File not found: path={file_or_directory}, " | ||||
|                     f"relative_url={__file_uri__}" | ||||
|                 ) | ||||
|                 raise not_found | ||||
|         return file_path | ||||
| @@ -1,7 +1,6 @@ | ||||
| from pathlib import Path | ||||
| from pathlib import PurePath | ||||
| from typing import Dict, Iterable, List, NamedTuple, Optional, Union | ||||
|  | ||||
| from sanic.handlers.directory import DirectoryHandler | ||||
| from sanic.models.handler_types import ( | ||||
|     ErrorMiddlewareType, | ||||
|     ListenerType, | ||||
| @@ -47,17 +46,16 @@ class FutureException(NamedTuple): | ||||
|  | ||||
| class FutureStatic(NamedTuple): | ||||
|     uri: str | ||||
|     file_or_directory: Path | ||||
|     file_or_directory: Union[str, bytes, PurePath] | ||||
|     pattern: str | ||||
|     use_modified_since: bool | ||||
|     use_content_range: bool | ||||
|     stream_large_files: Union[bool, int] | ||||
|     stream_large_files: bool | ||||
|     name: str | ||||
|     host: Optional[str] | ||||
|     strict_slashes: Optional[bool] | ||||
|     content_type: Optional[str] | ||||
|     content_type: Optional[bool] | ||||
|     resource_type: Optional[str] | ||||
|     directory_handler: DirectoryHandler | ||||
|  | ||||
|  | ||||
| class FutureSignal(NamedTuple): | ||||
|   | ||||
| @@ -1,51 +0,0 @@ | ||||
| from abc import ABC, abstractmethod | ||||
|  | ||||
| from html5tagger import HTML, Document | ||||
|  | ||||
| from sanic import __version__ as VERSION | ||||
| from sanic.application.logo import SVG_LOGO | ||||
| from sanic.pages.css import CSS | ||||
|  | ||||
|  | ||||
| class BasePage(ABC, metaclass=CSS):  # no cov | ||||
|     TITLE = "Unknown" | ||||
|     CSS: str | ||||
|  | ||||
|     def __init__(self, debug: bool = True) -> None: | ||||
|         self.doc = Document(self.TITLE, lang="en") | ||||
|         self.debug = debug | ||||
|  | ||||
|     @property | ||||
|     def style(self) -> str: | ||||
|         return self.CSS | ||||
|  | ||||
|     def render(self) -> str: | ||||
|         self._head() | ||||
|         self._body() | ||||
|         self._foot() | ||||
|         return str(self.doc) | ||||
|  | ||||
|     def _head(self) -> None: | ||||
|         self.doc.style(HTML(self.style)) | ||||
|         with self.doc.header: | ||||
|             self.doc.div(self.TITLE) | ||||
|  | ||||
|     def _foot(self) -> None: | ||||
|         with self.doc.footer: | ||||
|             self.doc.div("powered by") | ||||
|             with self.doc.div: | ||||
|                 self._sanic_logo() | ||||
|             if self.debug: | ||||
|                 self.doc.div(f"Version {VERSION}") | ||||
|  | ||||
|     @abstractmethod | ||||
|     def _body(self) -> None: | ||||
|         ... | ||||
|  | ||||
|     def _sanic_logo(self) -> None: | ||||
|         self.doc.a( | ||||
|             HTML(SVG_LOGO), | ||||
|             href="https://sanic.dev", | ||||
|             target="_blank", | ||||
|             referrerpolicy="no-referrer", | ||||
|         ) | ||||
| @@ -1,35 +0,0 @@ | ||||
| from abc import ABCMeta | ||||
| from pathlib import Path | ||||
| from typing import Optional | ||||
|  | ||||
|  | ||||
| CURRENT_DIR = Path(__file__).parent | ||||
|  | ||||
|  | ||||
| def _extract_style(maybe_style: Optional[str], name: str) -> str: | ||||
|     if maybe_style is not None: | ||||
|         maybe_path = Path(maybe_style) | ||||
|         if maybe_path.exists(): | ||||
|             return maybe_path.read_text(encoding="UTF-8") | ||||
|         return maybe_style | ||||
|     maybe_path = CURRENT_DIR / "styles" / f"{name}.css" | ||||
|     if maybe_path.exists(): | ||||
|         return maybe_path.read_text(encoding="UTF-8") | ||||
|     return "" | ||||
|  | ||||
|  | ||||
| class CSS(ABCMeta): | ||||
|     """Cascade stylesheets, i.e. combine all ancestor styles""" | ||||
|  | ||||
|     def __new__(cls, name, bases, attrs): | ||||
|         Page = super().__new__(cls, name, bases, attrs) | ||||
|         # Use a locally defined STYLE or the one from styles directory | ||||
|         s = _extract_style(attrs.get("STYLE"), name) | ||||
|         Page.STYLE = f"\n/* {name} */\n{s.strip()}\n" if s else "" | ||||
|         # Combine with all ancestor styles | ||||
|         Page.CSS = "".join( | ||||
|             Class.STYLE | ||||
|             for Class in reversed(Page.__mro__) | ||||
|             if type(Class) is CSS | ||||
|         ) | ||||
|         return Page | ||||
| @@ -1,66 +0,0 @@ | ||||
| import sys | ||||
|  | ||||
| from typing import Dict, Iterable | ||||
|  | ||||
| from html5tagger import E | ||||
|  | ||||
| from .base import BasePage | ||||
|  | ||||
|  | ||||
| if sys.version_info < (3, 8):  # no cov | ||||
|     FileInfo = Dict | ||||
|  | ||||
| else: | ||||
|     from typing import TypedDict | ||||
|  | ||||
|     class FileInfo(TypedDict): | ||||
|         icon: str | ||||
|         file_name: str | ||||
|         file_access: str | ||||
|         file_size: str | ||||
|  | ||||
|  | ||||
| class DirectoryPage(BasePage):  # no cov | ||||
|     TITLE = "Directory Viewer" | ||||
|  | ||||
|     def __init__( | ||||
|         self, files: Iterable[FileInfo], url: str, debug: bool | ||||
|     ) -> None: | ||||
|         super().__init__(debug) | ||||
|         self.files = files | ||||
|         self.url = url | ||||
|  | ||||
|     def _body(self) -> None: | ||||
|         with self.doc.main: | ||||
|             self._headline() | ||||
|             files = list(self.files) | ||||
|             if files: | ||||
|                 self._file_table(files) | ||||
|             else: | ||||
|                 self.doc.p("The folder is empty.") | ||||
|  | ||||
|     def _headline(self): | ||||
|         """Implement a heading with the current path, combined with | ||||
|         breadcrumb links""" | ||||
|         with self.doc.h1(id="breadcrumbs"): | ||||
|             p = self.url.split("/")[:-1] | ||||
|  | ||||
|             for i, part in enumerate(p): | ||||
|                 path = "/".join(p[: i + 1]) + "/" | ||||
|                 with self.doc.a(href=path): | ||||
|                     self.doc.span(part, class_="dir").span("/", class_="sep") | ||||
|  | ||||
|     def _file_table(self, files: Iterable[FileInfo]): | ||||
|         with self.doc.table(class_="autoindex container"): | ||||
|             for f in files: | ||||
|                 self._file_row(**f) | ||||
|  | ||||
|     def _file_row( | ||||
|         self, | ||||
|         icon: str, | ||||
|         file_name: str, | ||||
|         file_access: str, | ||||
|         file_size: str, | ||||
|     ): | ||||
|         first = E.span(icon, class_="icon").a(file_name, href=file_name) | ||||
|         self.doc.tr.td(first).td(file_size).td(file_access) | ||||
| @@ -1,79 +0,0 @@ | ||||
| html { | ||||
|     font: 16px sans-serif; | ||||
|     background: #eee; | ||||
|     color: #111; | ||||
| } | ||||
|  | ||||
| body { | ||||
|     margin: 0; | ||||
|     font-size: 1.25rem; | ||||
| } | ||||
|  | ||||
| body>* { | ||||
|     padding: 1rem 2vw; | ||||
| } | ||||
|  | ||||
| @media (max-width: 1200px) { | ||||
|     body>* { | ||||
|         padding: 0.5rem 1.5vw; | ||||
|     } | ||||
|  | ||||
|     body { | ||||
|         font-size: 1rem; | ||||
|     } | ||||
| } | ||||
|  | ||||
| .container { | ||||
|     min-width: 600px; | ||||
|     max-width: 1600px; | ||||
| } | ||||
|  | ||||
| header { | ||||
|     background: #111; | ||||
|     color: #e1e1e1; | ||||
|     border-bottom: 1px solid #272727; | ||||
|     text-align: center; | ||||
| } | ||||
|  | ||||
| footer { | ||||
|     text-align: center; | ||||
|     display: flex; | ||||
|     flex-direction: column; | ||||
|     font-size: 0.8rem; | ||||
|     margin-top: 2rem; | ||||
| } | ||||
|  | ||||
| h1 { | ||||
|     text-align: left; | ||||
| } | ||||
|  | ||||
| a:visited { | ||||
|     color: inherit; | ||||
| } | ||||
|  | ||||
| a { | ||||
|     text-decoration: none; | ||||
|     color: #88f; | ||||
| } | ||||
|  | ||||
| a:hover, | ||||
| a:focus { | ||||
|     text-decoration: underline; | ||||
|     outline: none; | ||||
| } | ||||
|  | ||||
| #logo { | ||||
|     height: 1.75rem; | ||||
|     padding: 0 0.25rem; | ||||
| } | ||||
|  | ||||
| span.icon { | ||||
|     margin-right: 1rem; | ||||
| } | ||||
|  | ||||
| @media (prefers-color-scheme: dark) { | ||||
|     html { | ||||
|         background: #111; | ||||
|         color: #ccc; | ||||
|     } | ||||
| } | ||||
| @@ -1,62 +0,0 @@ | ||||
| #breadcrumbs>a:hover { | ||||
|     text-decoration: none; | ||||
| } | ||||
|  | ||||
| #breadcrumbs>a .dir { | ||||
|     padding: 0 0.25em; | ||||
| } | ||||
|  | ||||
| #breadcrumbs>a:first-child:hover::before, | ||||
| #breadcrumbs>a .dir:hover { | ||||
|     text-decoration: underline; | ||||
| } | ||||
|  | ||||
| #breadcrumbs>a:first-child::before { | ||||
|     content: "🏠"; | ||||
| } | ||||
|  | ||||
| #breadcrumbs>a:last-child { | ||||
|     color: #ff0d68; | ||||
| } | ||||
|  | ||||
| main a { | ||||
|     color: inherit; | ||||
|     font-weight: bold; | ||||
| } | ||||
|  | ||||
| table.autoindex { | ||||
|     width: 100%; | ||||
|     font-family: monospace; | ||||
|     font-size: 1.25rem; | ||||
| } | ||||
|  | ||||
| table.autoindex tr { | ||||
|     display: flex; | ||||
| } | ||||
|  | ||||
| table.autoindex tr:hover { | ||||
|     background-color: #ddd; | ||||
| } | ||||
|  | ||||
| table.autoindex td { | ||||
|     margin: 0 0.5rem; | ||||
| } | ||||
|  | ||||
| table.autoindex td:first-child { | ||||
|     flex: 1; | ||||
| } | ||||
|  | ||||
| table.autoindex td:nth-child(2) { | ||||
|     text-align: right; | ||||
| } | ||||
|  | ||||
| table.autoindex td:last-child { | ||||
|     text-align: right; | ||||
| } | ||||
|  | ||||
|  | ||||
| @media (prefers-color-scheme: dark) { | ||||
|     table.autoindex tr:hover { | ||||
|         background-color: #222; | ||||
|     } | ||||
| } | ||||
| @@ -47,7 +47,7 @@ from sanic.constants import ( | ||||
| ) | ||||
| from sanic.exceptions import BadRequest, BadURL, ServerError | ||||
| from sanic.headers import ( | ||||
|     AcceptList, | ||||
|     AcceptContainer, | ||||
|     Options, | ||||
|     parse_accept, | ||||
|     parse_content_header, | ||||
| @@ -167,7 +167,7 @@ class Request: | ||||
|         self.conn_info: Optional[ConnInfo] = None | ||||
|         self.ctx = SimpleNamespace() | ||||
|         self.parsed_forwarded: Optional[Options] = None | ||||
|         self.parsed_accept: Optional[AcceptList] = None | ||||
|         self.parsed_accept: Optional[AcceptContainer] = None | ||||
|         self.parsed_credentials: Optional[Credentials] = None | ||||
|         self.parsed_json = None | ||||
|         self.parsed_form: Optional[RequestParameters] = None | ||||
| @@ -499,7 +499,7 @@ class Request: | ||||
|         return self.parsed_json | ||||
|  | ||||
|     @property | ||||
|     def accept(self) -> AcceptList: | ||||
|     def accept(self) -> AcceptContainer: | ||||
|         """ | ||||
|         :return: The ``Accept`` header parsed | ||||
|         :rtype: AcceptContainer | ||||
|   | ||||
| @@ -232,7 +232,7 @@ class JSONResponse(HTTPResponse): | ||||
|         body: Optional[Any] = None, | ||||
|         status: int = 200, | ||||
|         headers: Optional[Union[Header, Dict[str, str]]] = None, | ||||
|         content_type: Optional[str] = None, | ||||
|         content_type: str = "application/json", | ||||
|         dumps: Optional[Callable[..., str]] = None, | ||||
|         **kwargs: Any, | ||||
|     ): | ||||
|   | ||||
| @@ -45,7 +45,7 @@ class WebSocketConnection: | ||||
|  | ||||
|         await self._send(message) | ||||
|  | ||||
|     async def recv(self, *args, **kwargs) -> Optional[Union[str, bytes]]: | ||||
|     async def recv(self, *args, **kwargs) -> Optional[str]: | ||||
|         message = await self._receive() | ||||
|  | ||||
|         if message["type"] == "websocket.receive": | ||||
| @@ -53,7 +53,7 @@ class WebSocketConnection: | ||||
|                 return message["text"] | ||||
|             except KeyError: | ||||
|                 try: | ||||
|                     return message["bytes"] | ||||
|                     return message["bytes"].decode() | ||||
|                 except KeyError: | ||||
|                     raise InvalidUsage("Bad ASGI message received") | ||||
|         elif message["type"] == "websocket.disconnect": | ||||
|   | ||||
| @@ -2,6 +2,7 @@ from pathlib import Path | ||||
|  | ||||
| from sanic import Sanic | ||||
| from sanic.exceptions import SanicException | ||||
| from sanic.response import redirect | ||||
|  | ||||
|  | ||||
| def create_simple_server(directory: Path): | ||||
| @@ -11,8 +12,10 @@ def create_simple_server(directory: Path): | ||||
|         ) | ||||
|  | ||||
|     app = Sanic("SimpleServer") | ||||
|     app.static( | ||||
|         "/", directory, name="main", directory_view=True, index="index.html" | ||||
|     ) | ||||
|     app.static("/", directory, name="main") | ||||
|  | ||||
|     @app.get("/") | ||||
|     def index(_): | ||||
|         return redirect(app.url_for("main", filename="index.html")) | ||||
|  | ||||
|     return app | ||||
|   | ||||
							
								
								
									
										5
									
								
								setup.py
									
									
									
									
									
								
							
							
						
						
									
										5
									
								
								setup.py
									
									
									
									
									
								
							| @@ -35,7 +35,6 @@ def open_local(paths, mode="r", encoding="utf8"): | ||||
|  | ||||
|     return codecs.open(path, mode, encoding) | ||||
|  | ||||
|  | ||||
| def str_to_bool(val: str) -> bool: | ||||
|     val = val.lower() | ||||
|     if val in { | ||||
| @@ -56,7 +55,6 @@ def str_to_bool(val: str) -> bool: | ||||
|     else: | ||||
|         raise ValueError(f"Invalid truth value {val}") | ||||
|  | ||||
|  | ||||
| with open_local(["sanic", "__version__.py"], encoding="latin1") as fp: | ||||
|     try: | ||||
|         version = re.findall( | ||||
| @@ -81,7 +79,7 @@ setup_kwargs = { | ||||
|     ), | ||||
|     "long_description": long_description, | ||||
|     "packages": find_packages(exclude=("tests", "tests.*")), | ||||
|     "package_data": {"sanic": ["py.typed", "pages/styles/*"]}, | ||||
|     "package_data": {"sanic": ["py.typed"]}, | ||||
|     "platforms": "any", | ||||
|     "python_requires": ">=3.7", | ||||
|     "classifiers": [ | ||||
| @@ -111,7 +109,6 @@ requirements = [ | ||||
|     "aiofiles>=0.6.0", | ||||
|     "websockets>=10.0", | ||||
|     "multidict>=5.0,<7.0", | ||||
|     "html5tagger>=1.2.1", | ||||
| ] | ||||
|  | ||||
| tests_require = [ | ||||
|   | ||||
| @@ -1,7 +1,5 @@ | ||||
| import asyncio | ||||
| import inspect | ||||
| import logging | ||||
| import os | ||||
| import random | ||||
| import re | ||||
| import string | ||||
| @@ -60,6 +58,7 @@ CACHE: Dict[str, Any] = {} | ||||
|  | ||||
|  | ||||
| class RouteStringGenerator: | ||||
|  | ||||
|     ROUTE_COUNT_PER_DEPTH = 100 | ||||
|     HTTP_METHODS = HTTP_METHODS | ||||
|     ROUTE_PARAM_TYPES = ["str", "int", "float", "alpha", "uuid"] | ||||
| @@ -233,12 +232,3 @@ def urlopen(): | ||||
|     urlopen.read = Mock() | ||||
|     with patch("sanic.cli.inspector_client.urlopen", urlopen): | ||||
|         yield urlopen | ||||
|  | ||||
|  | ||||
| @pytest.fixture(scope="module") | ||||
| def static_file_directory(): | ||||
|     """The static directory to serve""" | ||||
|     current_file = inspect.getfile(inspect.currentframe()) | ||||
|     current_directory = os.path.dirname(os.path.abspath(current_file)) | ||||
|     static_directory = os.path.join(current_directory, "static") | ||||
|     return static_directory | ||||
|   | ||||
| @@ -36,7 +36,6 @@ def test_app_loop_running(app: Sanic): | ||||
|     assert response.text == "pass" | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_create_asyncio_server(app: Sanic): | ||||
|     loop = asyncio.get_event_loop() | ||||
|     asyncio_srv_coro = app.create_server(return_asyncio_server=True) | ||||
| @@ -45,7 +44,6 @@ def test_create_asyncio_server(app: Sanic): | ||||
|     assert srv.is_serving() is True | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_asyncio_server_no_start_serving(app: Sanic): | ||||
|     loop = asyncio.get_event_loop() | ||||
|     asyncio_srv_coro = app.create_server( | ||||
| @@ -57,7 +55,6 @@ def test_asyncio_server_no_start_serving(app: Sanic): | ||||
|     assert srv.is_serving() is False | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_asyncio_server_start_serving(app: Sanic): | ||||
|     loop = asyncio.get_event_loop() | ||||
|     asyncio_srv_coro = app.create_server( | ||||
| @@ -75,7 +72,6 @@ def test_asyncio_server_start_serving(app: Sanic): | ||||
|     # Looks like we can't easily test `serve_forever()` | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_create_server_main(app: Sanic, caplog): | ||||
|     app.listener("main_process_start")(lambda *_: ...) | ||||
|     loop = asyncio.get_event_loop() | ||||
| @@ -90,7 +86,6 @@ def test_create_server_main(app: Sanic, caplog): | ||||
|     ) in caplog.record_tuples | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_create_server_no_startup(app: Sanic): | ||||
|     loop = asyncio.get_event_loop() | ||||
|     asyncio_srv_coro = app.create_server( | ||||
| @@ -106,7 +101,6 @@ def test_create_server_no_startup(app: Sanic): | ||||
|         loop.run_until_complete(srv.start_serving()) | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_create_server_main_convenience(app: Sanic, caplog): | ||||
|     app.main_process_start(lambda *_: ...) | ||||
|     loop = asyncio.get_event_loop() | ||||
| @@ -132,6 +126,7 @@ def test_app_loop_not_running(app: Sanic): | ||||
|  | ||||
|  | ||||
| def test_app_run_raise_type_error(app: Sanic): | ||||
|  | ||||
|     with pytest.raises(TypeError) as excinfo: | ||||
|         app.run(loop="loop") | ||||
|  | ||||
| @@ -144,6 +139,7 @@ def test_app_run_raise_type_error(app: Sanic): | ||||
|  | ||||
|  | ||||
| def test_app_route_raise_value_error(app: Sanic): | ||||
|  | ||||
|     with pytest.raises(ValueError) as excinfo: | ||||
|  | ||||
|         @app.route("/test") | ||||
| @@ -225,6 +221,7 @@ def test_app_websocket_parameters(websocket_protocol_mock, app: Sanic): | ||||
|  | ||||
|  | ||||
| def test_handle_request_with_nested_exception(app: Sanic, monkeypatch): | ||||
|  | ||||
|     err_msg = "Mock Exception" | ||||
|  | ||||
|     def mock_error_handler_response(*args, **kwargs): | ||||
| @@ -244,6 +241,7 @@ def test_handle_request_with_nested_exception(app: Sanic, monkeypatch): | ||||
|  | ||||
|  | ||||
| def test_handle_request_with_nested_exception_debug(app: Sanic, monkeypatch): | ||||
|  | ||||
|     err_msg = "Mock Exception" | ||||
|  | ||||
|     def mock_error_handler_response(*args, **kwargs): | ||||
| @@ -472,7 +470,6 @@ def test_uvloop_config(app: Sanic, monkeypatch, use): | ||||
|         try_use_uvloop.assert_not_called() | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_uvloop_cannot_never_called_with_create_server(caplog, monkeypatch): | ||||
|     apps = (Sanic("default-uvloop"), Sanic("no-uvloop"), Sanic("yes-uvloop")) | ||||
|  | ||||
| @@ -509,7 +506,6 @@ def test_uvloop_cannot_never_called_with_create_server(caplog, monkeypatch): | ||||
|     assert counter[(logging.WARNING, message)] == modified | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| def test_multiple_uvloop_configs_display_warning(caplog): | ||||
|     Sanic._uvloop_setting = None  # Reset the setting (changed in prev tests) | ||||
|  | ||||
|   | ||||
| @@ -342,7 +342,7 @@ async def test_websocket_send(send, receive, message_stack): | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_websocket_text_receive(send, receive, message_stack): | ||||
| async def test_websocket_receive(send, receive, message_stack): | ||||
|     msg = {"text": "hello", "type": "websocket.receive"} | ||||
|     message_stack.append(msg) | ||||
|  | ||||
| @@ -351,15 +351,6 @@ async def test_websocket_text_receive(send, receive, message_stack): | ||||
|  | ||||
|     assert text == msg["text"] | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_websocket_bytes_receive(send, receive, message_stack): | ||||
|     msg = {"bytes": b"hello", "type": "websocket.receive"} | ||||
|     message_stack.append(msg) | ||||
|  | ||||
|     ws = WebSocketConnection(send, receive) | ||||
|     data = await ws.receive() | ||||
|  | ||||
|     assert data == msg["bytes"] | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_websocket_accept_with_no_subprotocols( | ||||
|   | ||||
| @@ -148,6 +148,7 @@ def test_cookie_set_unknown_property(): | ||||
|  | ||||
|  | ||||
| def test_cookie_set_same_key(app): | ||||
|  | ||||
|     cookies = {"test": "wait"} | ||||
|  | ||||
|     @app.get("/") | ||||
|   | ||||
| @@ -62,6 +62,7 @@ def exception_handler_app(): | ||||
|  | ||||
|     @exception_handler_app.route("/8", error_format="html") | ||||
|     def handler_8(request): | ||||
|  | ||||
|         raise ErrorWithRequestCtx("OK") | ||||
|  | ||||
|     @exception_handler_app.exception(ErrorWithRequestCtx, NotFound) | ||||
| @@ -213,7 +214,7 @@ def test_error_handler_noisy_log( | ||||
|     exception_handler_app: Sanic, monkeypatch: MonkeyPatch | ||||
| ): | ||||
|     err_logger = Mock() | ||||
|     monkeypatch.setattr(handlers.error, "error_logger", err_logger) | ||||
|     monkeypatch.setattr(handlers, "error_logger", err_logger) | ||||
|  | ||||
|     exception_handler_app.config["NOISY_EXCEPTIONS"] = False | ||||
|     exception_handler_app.test_client.get("/1") | ||||
|   | ||||
| @@ -185,22 +185,30 @@ def test_request_line(app): | ||||
|  | ||||
|     assert request.request_line == b"GET / HTTP/1.1" | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize( | ||||
|     "raw", | ||||
|     ( | ||||
|         "text/html, application/xhtml+xml, application/xml;q=0.9, */*;q=0.8", | ||||
|         "application/xml;q=0.9, */*;q=0.8, text/html, application/xhtml+xml", | ||||
|         "foo/bar;q=0.9, */*;q=0.8, text/html=0.8, text/plain, application/xhtml+xml", | ||||
|         "show/first, show/second", | ||||
|         "show/*, show/first", | ||||
|         "*/*, show/first", | ||||
|         "*/*, show/*", | ||||
|         "other/*; q=0.1, show/*; q=0.2", | ||||
|         "show/first; q=0.5, show/second; q=0.5", | ||||
|         "show/first; foo=bar, show/second; foo=bar", | ||||
|         "show/second, show/first; foo=bar", | ||||
|         "show/second; q=0.5, show/first; foo=bar; q=0.5", | ||||
|         "show/second; q=0.5, show/first; q=1.0", | ||||
|         "show/first, show/second; q=1.0", | ||||
|     ), | ||||
| ) | ||||
| def test_parse_accept_ordered_okay(raw): | ||||
|     ordered = headers.parse_accept(raw) | ||||
|     expected_subtype = ( | ||||
|         "*" if all(q.subtype.is_wildcard for q in ordered) else "first" | ||||
|     ) | ||||
| def test_accept_ordering(raw): | ||||
|     """Should sort by q but also be stable.""" | ||||
|     accept = headers.parse_accept(raw) | ||||
|     assert accept[0].type_ == "text" | ||||
|     raw1 =  ", ".join(str(a) for a in accept) | ||||
|     accept = headers.parse_accept(raw1) | ||||
|     raw2 =  ", ".join(str(a) for a in accept) | ||||
|     assert raw1 == raw2 | ||||
|     assert ordered[0].type_ == "show" | ||||
|     assert ordered[0].subtype == expected_subtype | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize( | ||||
| @@ -217,27 +225,40 @@ def test_bad_accept(raw): | ||||
|  | ||||
|  | ||||
| def test_empty_accept(): | ||||
|     a = headers.parse_accept("") | ||||
|     assert a == [] | ||||
|     assert not a.match("*/*") | ||||
|     assert headers.parse_accept("") == [] | ||||
|  | ||||
|  | ||||
| def test_wildcard_accept_set_ok(): | ||||
|     accept = headers.parse_accept("*/*")[0] | ||||
|     assert accept.is_wildcard | ||||
|     assert accept.has_wildcard | ||||
|  | ||||
|     accept = headers.parse_accept("foo/*")[0] | ||||
|     assert not accept.is_wildcard | ||||
|     assert accept.has_wildcard | ||||
|  | ||||
|     accept = headers.parse_accept("*/bar")[0] | ||||
|     assert not accept.is_wildcard | ||||
|     assert accept.has_wildcard | ||||
|     assert accept.type_.is_wildcard | ||||
|     assert accept.subtype.is_wildcard | ||||
|  | ||||
|     accept = headers.parse_accept("foo/bar")[0] | ||||
|     assert not accept.is_wildcard | ||||
|     assert not accept.has_wildcard | ||||
|     assert not accept.type_.is_wildcard | ||||
|     assert not accept.subtype.is_wildcard | ||||
|  | ||||
|  | ||||
| def test_accept_parsed_against_str(): | ||||
|     accept = headers.Accept.parse("foo/bar") | ||||
|     assert accept > "foo/bar; q=0.1" | ||||
|  | ||||
|  | ||||
| def test_media_type_equality(): | ||||
|     assert headers.MediaType("foo") == headers.MediaType("foo") == "foo" | ||||
|     assert headers.MediaType("foo") == headers.MediaType("*") == "*" | ||||
|     assert headers.MediaType("foo") != headers.MediaType("bar") | ||||
|     assert headers.MediaType("foo") != "bar" | ||||
|  | ||||
|  | ||||
| def test_media_type_matching(): | ||||
|     assert headers.MediaType("foo").match(headers.MediaType("foo")) | ||||
|     assert headers.MediaType("foo").match("foo") | ||||
|  | ||||
|     assert not headers.MediaType("foo").match(headers.MediaType("*")) | ||||
|     assert not headers.MediaType("foo").match("*") | ||||
|  | ||||
|     assert not headers.MediaType("foo").match(headers.MediaType("bar")) | ||||
|     assert not headers.MediaType("foo").match("bar") | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize( | ||||
| @@ -245,52 +266,87 @@ def test_wildcard_accept_set_ok(): | ||||
|     ( | ||||
|         # ALLOW BOTH | ||||
|         ("foo/bar", "foo/bar", True, True, True), | ||||
|         ("foo/bar", headers.Accept.parse("foo/bar"), True, True, True), | ||||
|         ("foo/bar", "foo/*", True, True, True), | ||||
|         ("foo/bar", headers.Accept.parse("foo/*"), True, True, True), | ||||
|         ("foo/bar", "*/*", True, True, True), | ||||
|         ("foo/bar", headers.Accept.parse("*/*"), True, True, True), | ||||
|         ("foo/*", "foo/bar", True, True, True), | ||||
|         ("foo/*", headers.Accept.parse("foo/bar"), True, True, True), | ||||
|         ("foo/*", "foo/*", True, True, True), | ||||
|         ("foo/*", headers.Accept.parse("foo/*"), True, True, True), | ||||
|         ("foo/*", "*/*", True, True, True), | ||||
|         ("foo/*", headers.Accept.parse("*/*"), True, True, True), | ||||
|         ("*/*", "foo/bar", True, True, True), | ||||
|         ("*/*", headers.Accept.parse("foo/bar"), True, True, True), | ||||
|         ("*/*", "foo/*", True, True, True), | ||||
|         ("*/*", headers.Accept.parse("foo/*"), True, True, True), | ||||
|         ("*/*", "*/*", True, True, True), | ||||
|         ("*/*", headers.Accept.parse("*/*"), True, True, True), | ||||
|         # ALLOW TYPE | ||||
|         ("foo/bar", "foo/bar", True, True, False), | ||||
|         ("foo/bar", headers.Accept.parse("foo/bar"), True, True, False), | ||||
|         ("foo/bar", "foo/*", False, True, False), | ||||
|         ("foo/bar", headers.Accept.parse("foo/*"), False, True, False), | ||||
|         ("foo/bar", "*/*", False, True, False), | ||||
|         ("foo/bar", headers.Accept.parse("*/*"), False, True, False), | ||||
|         ("foo/*", "foo/bar", False, True, False), | ||||
|         ("foo/*", headers.Accept.parse("foo/bar"), False, True, False), | ||||
|         ("foo/*", "foo/*", False, True, False), | ||||
|         ("foo/*", headers.Accept.parse("foo/*"), False, True, False), | ||||
|         ("foo/*", "*/*", False, True, False), | ||||
|         ("foo/*", headers.Accept.parse("*/*"), False, True, False), | ||||
|         ("*/*", "foo/bar", False, True, False), | ||||
|         ("*/*", headers.Accept.parse("foo/bar"), False, True, False), | ||||
|         ("*/*", "foo/*", False, True, False), | ||||
|         ("*/*", headers.Accept.parse("foo/*"), False, True, False), | ||||
|         ("*/*", "*/*", False, True, False), | ||||
|         ("*/*", headers.Accept.parse("*/*"), False, True, False), | ||||
|         # ALLOW SUBTYPE | ||||
|         ("foo/bar", "foo/bar", True, False, True), | ||||
|         ("foo/bar", headers.Accept.parse("foo/bar"), True, False, True), | ||||
|         ("foo/bar", "foo/*", True, False, True), | ||||
|         ("foo/bar", headers.Accept.parse("foo/*"), True, False, True), | ||||
|         ("foo/bar", "*/*", False, False, True), | ||||
|         ("foo/bar", headers.Accept.parse("*/*"), False, False, True), | ||||
|         ("foo/*", "foo/bar", True, False, True), | ||||
|         ("foo/*", headers.Accept.parse("foo/bar"), True, False, True), | ||||
|         ("foo/*", "foo/*", True, False, True), | ||||
|         ("foo/*", headers.Accept.parse("foo/*"), True, False, True), | ||||
|         ("foo/*", "*/*", False, False, True), | ||||
|         ("foo/*", headers.Accept.parse("*/*"), False, False, True), | ||||
|         ("*/*", "foo/bar", False, False, True), | ||||
|         ("*/*", headers.Accept.parse("foo/bar"), False, False, True), | ||||
|         ("*/*", "foo/*", False, False, True), | ||||
|         ("*/*", headers.Accept.parse("foo/*"), False, False, True), | ||||
|         ("*/*", "*/*", False, False, True), | ||||
|         ("*/*", headers.Accept.parse("*/*"), False, False, True), | ||||
|     ), | ||||
| ) | ||||
| def test_accept_matching(value, other, outcome, allow_type, allow_subtype): | ||||
|     assert ( | ||||
|         bool(headers.MediaType._parse(value).match( | ||||
|         headers.Accept.parse(value).match( | ||||
|             other, | ||||
|             allow_type_wildcard=allow_type, | ||||
|             allow_subtype_wildcard=allow_subtype, | ||||
|         )) | ||||
|         ) | ||||
|         is outcome | ||||
|     ) | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize("value", ("foo/bar", "foo/*", "*/*")) | ||||
| def test_value_in_accept(value): | ||||
|     acceptable = headers.parse_accept(value) | ||||
|     assert "foo/bar" in acceptable | ||||
|     assert "foo/*" in acceptable | ||||
|     assert "*/*" in acceptable | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize("value", ("foo/bar", "foo/*")) | ||||
| def test_value_not_in_accept(value): | ||||
|     acceptable = headers.parse_accept(value) | ||||
|     assert "*/*" not in acceptable | ||||
|     assert "*/bar" not in acceptable | ||||
|     assert "no/match" not in acceptable | ||||
|     assert "no/*" not in acceptable | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize( | ||||
| @@ -299,22 +355,16 @@ def test_value_not_in_accept(value): | ||||
|         ( | ||||
|             "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",  # noqa: E501 | ||||
|             [ | ||||
|                 ("text/html", 1.0), | ||||
|                 ("application/xhtml+xml", 1.0), | ||||
|                 ("image/avif", 1.0), | ||||
|                 ("image/webp", 1.0), | ||||
|                 ("application/xml", 0.9), | ||||
|                 ("*/*", 0.8), | ||||
|                 "text/html", | ||||
|                 "application/xhtml+xml", | ||||
|                 "image/avif", | ||||
|                 "image/webp", | ||||
|                 "application/xml;q=0.9", | ||||
|                 "*/*;q=0.8", | ||||
|             ], | ||||
|         ), | ||||
|     ), | ||||
| ) | ||||
| def test_browser_headers(header, expected): | ||||
|     mimes = [e[0] for e in expected] | ||||
|     qs = [e[1] for e in expected] | ||||
|     request = Request(b"/", {"accept": header}, "1.1", "GET", None, None) | ||||
|     assert request.accept == mimes | ||||
|     for a, m, q in zip(request.accept, mimes, qs): | ||||
|         assert a == m | ||||
|         assert a.str == m | ||||
|         assert a.q == q | ||||
|     assert request.accept == expected | ||||
|   | ||||
| @@ -514,6 +514,7 @@ def test_file_stream_head_response( | ||||
| def test_file_stream_response_range( | ||||
|     app: Sanic, file_name, static_file_directory, size, start, end | ||||
| ): | ||||
|  | ||||
|     Range = namedtuple("Range", ["size", "start", "end", "total"]) | ||||
|     total = len(get_file_content(static_file_directory, file_name)) | ||||
|     range = Range(size=size, start=start, end=end, total=total) | ||||
|   | ||||
| @@ -213,3 +213,12 @@ def test_pop_list(json_app: Sanic): | ||||
|  | ||||
|     _, resp = json_app.test_client.get("/json-pop") | ||||
|     assert resp.body == json_dumps(["b"]).encode() | ||||
|  | ||||
|  | ||||
| def test_json_response_class_sets_proper_content_type(json_app: Sanic): | ||||
|     @json_app.get("/json-class") | ||||
|     async def handler(request: Request): | ||||
|         return JSONResponse(JSON_BODY) | ||||
|  | ||||
|     _, resp = json_app.test_client.get("/json-class") | ||||
|     assert resp.headers["content-type"] == "application/json" | ||||
|   | ||||
| @@ -722,6 +722,7 @@ def test_add_webscoket_route_with_version(app): | ||||
|  | ||||
|  | ||||
| def test_route_duplicate(app): | ||||
|  | ||||
|     with pytest.raises(RouteExists): | ||||
|  | ||||
|         @app.route("/test") | ||||
| @@ -818,6 +819,7 @@ def test_unquote_add_route(app, unquote): | ||||
|  | ||||
|  | ||||
| def test_dynamic_add_route(app): | ||||
|  | ||||
|     results = [] | ||||
|  | ||||
|     async def handler(request, name): | ||||
| @@ -832,6 +834,7 @@ def test_dynamic_add_route(app): | ||||
|  | ||||
|  | ||||
| def test_dynamic_add_route_string(app): | ||||
|  | ||||
|     results = [] | ||||
|  | ||||
|     async def handler(request, name): | ||||
| @@ -935,6 +938,7 @@ def test_dynamic_add_route_unhashable(app): | ||||
|  | ||||
|  | ||||
| def test_add_route_duplicate(app): | ||||
|  | ||||
|     with pytest.raises(RouteExists): | ||||
|  | ||||
|         async def handler1(request): | ||||
| @@ -1116,6 +1120,7 @@ def test_route_raise_ParameterNameConflicts(app): | ||||
|  | ||||
|  | ||||
| def test_route_invalid_host(app): | ||||
|  | ||||
|     host = 321 | ||||
|     with pytest.raises(ValueError) as excinfo: | ||||
|  | ||||
|   | ||||
| @@ -93,7 +93,6 @@ def test_dont_register_system_signals(app): | ||||
| @pytest.mark.skipif(os.name == "nt", reason="windows cannot SIGINT processes") | ||||
| def test_windows_workaround(): | ||||
|     """Test Windows workaround (on any other OS)""" | ||||
|  | ||||
|     # At least some code coverage, even though this test doesn't work on | ||||
|     # Windows... | ||||
|     class MockApp: | ||||
|   | ||||
| @@ -1,3 +1,4 @@ | ||||
| import inspect | ||||
| import logging | ||||
| import os | ||||
| import sys | ||||
| @@ -12,6 +13,15 @@ from sanic import Sanic, text | ||||
| from sanic.exceptions import FileNotFound | ||||
|  | ||||
|  | ||||
| @pytest.fixture(scope="module") | ||||
| def static_file_directory(): | ||||
|     """The static directory to serve""" | ||||
|     current_file = inspect.getfile(inspect.currentframe()) | ||||
|     current_directory = os.path.dirname(os.path.abspath(current_file)) | ||||
|     static_directory = os.path.join(current_directory, "static") | ||||
|     return static_directory | ||||
|  | ||||
|  | ||||
| @pytest.fixture(scope="module") | ||||
| def double_dotted_directory_file(static_file_directory: str): | ||||
|     """Generate double dotted directory and its files""" | ||||
| @@ -108,11 +118,6 @@ def test_static_file_pathlib(app, static_file_directory, file_name): | ||||
| def test_static_file_bytes(app, static_file_directory, file_name): | ||||
|     bsep = os.path.sep.encode("utf-8") | ||||
|     file_path = static_file_directory.encode("utf-8") + bsep + file_name | ||||
|     message = ( | ||||
|         "Serving a static directory with a bytes " | ||||
|         "string is deprecated and will be removed in v22.9." | ||||
|     ) | ||||
|     with pytest.warns(DeprecationWarning, match=message): | ||||
|     app.static("/testing.file", file_path) | ||||
|     request, response = app.test_client.get("/testing.file") | ||||
|     assert response.status == 200 | ||||
| @@ -426,6 +431,7 @@ def test_static_stream_large_file( | ||||
|     "file_name", ["test.file", "decode me.txt", "python.png"] | ||||
| ) | ||||
| def test_use_modified_since(app, static_file_directory, file_name): | ||||
|  | ||||
|     file_stat = os.stat(get_file_path(static_file_directory, file_name)) | ||||
|     modified_since = strftime( | ||||
|         "%a, %d %b %Y %H:%M:%S GMT", gmtime(file_stat.st_mtime) | ||||
|   | ||||
| @@ -1,123 +0,0 @@ | ||||
| import os | ||||
|  | ||||
| from pathlib import Path | ||||
|  | ||||
| import pytest | ||||
|  | ||||
| from sanic import Sanic | ||||
| from sanic.handlers.directory import DirectoryHandler | ||||
|  | ||||
|  | ||||
| def get_file_path(static_file_directory, file_name): | ||||
|     return os.path.join(static_file_directory, file_name) | ||||
|  | ||||
|  | ||||
| def get_file_content(static_file_directory, file_name): | ||||
|     """The content of the static file to check""" | ||||
|     with open(get_file_path(static_file_directory, file_name), "rb") as file: | ||||
|         return file.read() | ||||
|  | ||||
|  | ||||
| def test_static_directory_view(app: Sanic, static_file_directory: str): | ||||
|     app.static("/static", static_file_directory, directory_view=True) | ||||
|  | ||||
|     _, response = app.test_client.get("/static/") | ||||
|     assert response.status == 200 | ||||
|     assert response.content_type == "text/html; charset=utf-8" | ||||
|     assert "<title>Directory Viewer</title>" in response.text | ||||
|  | ||||
|  | ||||
| def test_static_index_single(app: Sanic, static_file_directory: str): | ||||
|     app.static("/static", static_file_directory, index="test.html") | ||||
|  | ||||
|     _, response = app.test_client.get("/static/") | ||||
|     assert response.status == 200 | ||||
|     assert response.body == get_file_content( | ||||
|         static_file_directory, "test.html" | ||||
|     ) | ||||
|     assert response.headers["Content-Type"] == "text/html" | ||||
|  | ||||
|  | ||||
| def test_static_index_single_not_found(app: Sanic, static_file_directory: str): | ||||
|     app.static("/static", static_file_directory, index="index.html") | ||||
|  | ||||
|     _, response = app.test_client.get("/static/") | ||||
|     assert response.status == 404 | ||||
|  | ||||
|  | ||||
| def test_static_index_multiple(app: Sanic, static_file_directory: str): | ||||
|     app.static( | ||||
|         "/static", | ||||
|         static_file_directory, | ||||
|         index=["index.html", "test.html"], | ||||
|     ) | ||||
|  | ||||
|     _, response = app.test_client.get("/static/") | ||||
|     assert response.status == 200 | ||||
|     assert response.body == get_file_content( | ||||
|         static_file_directory, "test.html" | ||||
|     ) | ||||
|     assert response.headers["Content-Type"] == "text/html" | ||||
|  | ||||
|  | ||||
| def test_static_directory_view_and_index( | ||||
|     app: Sanic, static_file_directory: str | ||||
| ): | ||||
|     app.static( | ||||
|         "/static", | ||||
|         static_file_directory, | ||||
|         directory_view=True, | ||||
|         index="foo.txt", | ||||
|     ) | ||||
|  | ||||
|     _, response = app.test_client.get("/static/nested/") | ||||
|     assert response.status == 200 | ||||
|     assert response.content_type == "text/html; charset=utf-8" | ||||
|     assert "<title>Directory Viewer</title>" in response.text | ||||
|  | ||||
|     _, response = app.test_client.get("/static/nested/dir/") | ||||
|     assert response.status == 200 | ||||
|     assert response.body == get_file_content( | ||||
|         f"{static_file_directory}/nested/dir", "foo.txt" | ||||
|     ) | ||||
|     assert response.content_type == "text/plain" | ||||
|  | ||||
|  | ||||
| def test_static_directory_handler(app: Sanic, static_file_directory: str): | ||||
|     dh = DirectoryHandler( | ||||
|         "/static", | ||||
|         Path(static_file_directory), | ||||
|         directory_view=True, | ||||
|         index="foo.txt", | ||||
|     ) | ||||
|     app.static("/static", static_file_directory, directory_handler=dh) | ||||
|  | ||||
|     _, response = app.test_client.get("/static/nested/") | ||||
|     assert response.status == 200 | ||||
|     assert response.content_type == "text/html; charset=utf-8" | ||||
|     assert "<title>Directory Viewer</title>" in response.text | ||||
|  | ||||
|     _, response = app.test_client.get("/static/nested/dir/") | ||||
|     assert response.status == 200 | ||||
|     assert response.body == get_file_content( | ||||
|         f"{static_file_directory}/nested/dir", "foo.txt" | ||||
|     ) | ||||
|     assert response.content_type == "text/plain" | ||||
|  | ||||
|  | ||||
| def test_static_directory_handler_fails(app: Sanic): | ||||
|     dh = DirectoryHandler( | ||||
|         "/static", | ||||
|         Path(""), | ||||
|         directory_view=True, | ||||
|         index="foo.txt", | ||||
|     ) | ||||
|     message = ( | ||||
|         "When explicitly setting directory_handler, you cannot " | ||||
|         "set either directory_view or index. Instead, pass " | ||||
|         "these arguments to your DirectoryHandler instance." | ||||
|     ) | ||||
|     with pytest.raises(ValueError, match=message): | ||||
|         app.static("/static", "", directory_handler=dh, directory_view=True) | ||||
|     with pytest.raises(ValueError, match=message): | ||||
|         app.static("/static", "", directory_handler=dh, index="index.html") | ||||
| @@ -654,6 +654,7 @@ def test_sanic_ssl_context_create(): | ||||
|     reason="This test requires fork context", | ||||
| ) | ||||
| def test_ssl_in_multiprocess_mode(app: Sanic, caplog): | ||||
|  | ||||
|     ssl_dict = {"cert": localhost_cert, "key": localhost_key} | ||||
|     event = Event() | ||||
|  | ||||
|   | ||||
| @@ -176,6 +176,7 @@ def handler(request: Request): | ||||
|  | ||||
| async def client(app: Sanic, loop: AbstractEventLoop): | ||||
|     try: | ||||
|  | ||||
|         transport = httpx.AsyncHTTPTransport(uds=SOCKPATH) | ||||
|         async with httpx.AsyncClient(transport=transport) as client: | ||||
|             r = await client.get("http://myhost.invalid/") | ||||
|   | ||||
| @@ -83,6 +83,7 @@ def test_simple_url_for_getting_with_more_params(app, args, url): | ||||
|  | ||||
|  | ||||
| def test_url_for_with_server_name(app): | ||||
|  | ||||
|     server_name = f"{test_host}:{test_port}" | ||||
|     app.config.update({"SERVER_NAME": server_name}) | ||||
|     path = "/myurl" | ||||
|   | ||||
| @@ -38,6 +38,7 @@ def test_load_module_from_file_location_with_non_existing_env_variable(): | ||||
|         LoadFileException, | ||||
|         match="The following environment variables are not set: MuuMilk", | ||||
|     ): | ||||
|  | ||||
|         load_module_from_file_location("${MuuMilk}") | ||||
|  | ||||
|  | ||||
|   | ||||
		Reference in New Issue
	
	Block a user