6c03dd87b1
* Allow Pathlib Path objects to be passed to the app.static file endpoint register helper. * fixed import sort * Raise error if static file path is not an accepted object type Added more tests to improve coverage on the new type checks.
190 lines
6.6 KiB
Python
190 lines
6.6 KiB
Python
from functools import partial, wraps
|
|
from mimetypes import guess_type
|
|
from os import path
|
|
from pathlib import PurePath
|
|
from re import sub
|
|
from time import gmtime, strftime
|
|
from typing import Union
|
|
from urllib.parse import unquote
|
|
|
|
from sanic.compat import stat_async
|
|
from sanic.exceptions import (
|
|
ContentRangeError,
|
|
FileNotFound,
|
|
HeaderNotFound,
|
|
InvalidUsage,
|
|
)
|
|
from sanic.handlers import ContentRangeHandler
|
|
from sanic.log import error_logger
|
|
from sanic.response import HTTPResponse, file, file_stream
|
|
|
|
|
|
async def _static_request_handler(
|
|
file_or_directory,
|
|
use_modified_since,
|
|
use_content_range,
|
|
stream_large_files,
|
|
request,
|
|
content_type=None,
|
|
file_uri=None,
|
|
):
|
|
# Using this to determine if the URL is trying to break out of the path
|
|
# served. os.path.realpath seems to be very slow
|
|
if file_uri and "../" in file_uri:
|
|
raise InvalidUsage("Invalid URL")
|
|
# Merge served directory and requested file if provided
|
|
# 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
|
|
root_path = file_path = file_or_directory
|
|
if file_uri:
|
|
file_path = path.join(file_or_directory, sub("^[/]*", "", file_uri))
|
|
|
|
# URL decode the path sent by the browser otherwise we won't be able to
|
|
# match filenames which got encoded (filenames with spaces etc)
|
|
file_path = path.abspath(unquote(file_path))
|
|
if not file_path.startswith(path.abspath(unquote(root_path))):
|
|
error_logger.exception(
|
|
f"File not found: path={file_or_directory}, "
|
|
f"relative_url={file_uri}"
|
|
)
|
|
raise FileNotFound(
|
|
"File not found", path=file_or_directory, relative_url=file_uri
|
|
)
|
|
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 = strftime(
|
|
"%a, %d %b %Y %H:%M:%S GMT", gmtime(stats.st_mtime)
|
|
)
|
|
if request.headers.get("If-Modified-Since") == modified_since:
|
|
return HTTPResponse(status=304)
|
|
headers["Last-Modified"] = modified_since
|
|
_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"]
|
|
for key, value in _range.headers.items():
|
|
headers[key] = value
|
|
headers["Content-Type"] = (
|
|
content_type or guess_type(file_path)[0] or "text/plain"
|
|
)
|
|
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 ContentRangeError:
|
|
raise
|
|
except Exception:
|
|
error_logger.exception(
|
|
f"File not found: path={file_or_directory}, "
|
|
f"relative_url={file_uri}"
|
|
)
|
|
raise FileNotFound(
|
|
"File not found", path=file_or_directory, relative_url=file_uri
|
|
)
|
|
|
|
|
|
def register(
|
|
app,
|
|
uri: str,
|
|
file_or_directory: Union[str, bytes, PurePath],
|
|
pattern,
|
|
use_modified_since,
|
|
use_content_range,
|
|
stream_large_files,
|
|
name: str = "static",
|
|
host=None,
|
|
strict_slashes=None,
|
|
content_type=None,
|
|
):
|
|
# 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(file_or_directory, bytes):
|
|
file_or_directory = file_or_directory.decode("utf-8")
|
|
elif isinstance(file_or_directory, PurePath):
|
|
file_or_directory = str(file_or_directory)
|
|
elif not isinstance(file_or_directory, str):
|
|
raise ValueError("Invalid file path string.")
|
|
|
|
# If we're not trying to match a file directly,
|
|
# serve from the folder
|
|
if not path.isfile(file_or_directory):
|
|
uri += "<file_uri:" + pattern + ">"
|
|
|
|
# special prefix for static files
|
|
if not name.startswith("_static_"):
|
|
name = f"_static_{name}"
|
|
|
|
_handler = wraps(_static_request_handler)(
|
|
partial(
|
|
_static_request_handler,
|
|
file_or_directory,
|
|
use_modified_since,
|
|
use_content_range,
|
|
stream_large_files,
|
|
content_type=content_type,
|
|
)
|
|
)
|
|
|
|
_routes, _ = app.route(
|
|
uri,
|
|
methods=["GET", "HEAD"],
|
|
name=name,
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
)(_handler)
|
|
return _routes
|