From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../site-packages/litestar/response/file.py | 386 +++++++++++++++++++++ 1 file changed, 386 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/response/file.py (limited to 'venv/lib/python3.11/site-packages/litestar/response/file.py') diff --git a/venv/lib/python3.11/site-packages/litestar/response/file.py b/venv/lib/python3.11/site-packages/litestar/response/file.py new file mode 100644 index 0000000..1fc6f86 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/response/file.py @@ -0,0 +1,386 @@ +from __future__ import annotations + +import itertools +from email.utils import formatdate +from inspect import iscoroutine +from mimetypes import encodings_map, guess_type +from typing import TYPE_CHECKING, Any, AsyncGenerator, Coroutine, Iterable, Literal, cast +from urllib.parse import quote +from zlib import adler32 + +from litestar.constants import ONE_MEGABYTE +from litestar.exceptions import ImproperlyConfiguredException +from litestar.file_system import BaseLocalFileSystem, FileSystemAdapter +from litestar.response.base import Response +from litestar.response.streaming import ASGIStreamingResponse +from litestar.utils.deprecation import warn_deprecation +from litestar.utils.helpers import get_enum_string_value + +if TYPE_CHECKING: + from os import PathLike + from os import stat_result as stat_result_type + + from anyio import Path + + from litestar.app import Litestar + from litestar.background_tasks import BackgroundTask, BackgroundTasks + from litestar.connection import Request + from litestar.datastructures.cookie import Cookie + from litestar.datastructures.headers import ETag + from litestar.enums import MediaType + from litestar.types import ( + HTTPResponseBodyEvent, + PathType, + Receive, + ResponseCookies, + ResponseHeaders, + Send, + TypeEncodersMap, + ) + from litestar.types.file_types import FileInfo, FileSystemProtocol + +__all__ = ( + "ASGIFileResponse", + "File", + "async_file_iterator", + "create_etag_for_file", +) + +# brotli not supported in 'mimetypes.encodings_map' until py 3.9. +encodings_map[".br"] = "br" + + +async def async_file_iterator( + file_path: PathType, chunk_size: int, adapter: FileSystemAdapter +) -> AsyncGenerator[bytes, None]: + """Return an async that asynchronously reads a file and yields its chunks. + + Args: + file_path: A path to a file. + chunk_size: The chunk file to use. + adapter: File system adapter class. + adapter: File system adapter class. + + Returns: + An async generator. + """ + async with await adapter.open(file_path) as file: + while chunk := await file.read(chunk_size): + yield chunk + + +def create_etag_for_file(path: PathType, modified_time: float, file_size: int) -> str: + """Create an etag. + + Notes: + - Function is derived from flask. + + Returns: + An etag. + """ + check = adler32(str(path).encode("utf-8")) & 0xFFFFFFFF + return f'"{modified_time}-{file_size}-{check}"' + + +class ASGIFileResponse(ASGIStreamingResponse): + """A low-level ASGI response, streaming a file as response body.""" + + def __init__( + self, + *, + background: BackgroundTask | BackgroundTasks | None = None, + body: bytes | str = b"", + chunk_size: int = ONE_MEGABYTE, + content_disposition_type: Literal["attachment", "inline"] = "attachment", + content_length: int | None = None, + cookies: Iterable[Cookie] | None = None, + encoded_headers: Iterable[tuple[bytes, bytes]] | None = None, + encoding: str = "utf-8", + etag: ETag | None = None, + file_info: FileInfo | Coroutine[None, None, FileInfo] | None = None, + file_path: str | PathLike | Path, + file_system: FileSystemProtocol | None = None, + filename: str = "", + headers: dict[str, str] | None = None, + is_head_response: bool = False, + media_type: MediaType | str | None = None, + stat_result: stat_result_type | None = None, + status_code: int | None = None, + ) -> None: + """A low-level ASGI response, streaming a file as response body. + + Args: + background: A background task or a list of background tasks to be executed after the response is sent. + body: encoded content to send in the response body. + chunk_size: The chunk size to use. + content_disposition_type: The type of the ``Content-Disposition``. Either ``inline`` or ``attachment``. + content_length: The response content length. + cookies: The response cookies. + encoded_headers: A list of encoded headers. + encoding: The response encoding. + etag: An etag. + file_info: A file info. + file_path: A path to a file. + file_system: A file system adapter. + filename: The name of the file. + headers: A dictionary of headers. + headers: The response headers. + is_head_response: A boolean indicating if the response is a HEAD response. + media_type: The media type of the file. + stat_result: A stat result. + status_code: The response status code. + """ + headers = headers or {} + if not media_type: + mimetype, content_encoding = guess_type(filename) if filename else (None, None) + media_type = mimetype or "application/octet-stream" + if content_encoding is not None: + headers.update({"content-encoding": content_encoding}) + + self.adapter = FileSystemAdapter(file_system or BaseLocalFileSystem()) + + super().__init__( + iterator=async_file_iterator(file_path=file_path, chunk_size=chunk_size, adapter=self.adapter), + headers=headers, + media_type=media_type, + cookies=cookies, + background=background, + status_code=status_code, + body=body, + content_length=content_length, + encoding=encoding, + is_head_response=is_head_response, + encoded_headers=encoded_headers, + ) + + quoted_filename = quote(filename) + is_utf8 = quoted_filename == filename + if is_utf8: + content_disposition = f'{content_disposition_type}; filename="{filename}"' + else: + content_disposition = f"{content_disposition_type}; filename*=utf-8''{quoted_filename}" + + self.headers.setdefault("content-disposition", content_disposition) + + self.chunk_size = chunk_size + self.etag = etag + self.file_path = file_path + + if file_info: + self.file_info: FileInfo | Coroutine[Any, Any, FileInfo] = file_info + elif stat_result: + self.file_info = self.adapter.parse_stat_result(result=stat_result, path=file_path) + else: + self.file_info = self.adapter.info(self.file_path) + + async def send_body(self, send: Send, receive: Receive) -> None: + """Emit a stream of events correlating with the response body. + + Args: + send: The ASGI send function. + receive: The ASGI receive function. + + Returns: + None + """ + if self.chunk_size < self.content_length: + await super().send_body(send=send, receive=receive) + return + + async with await self.adapter.open(self.file_path) as file: + body_event: HTTPResponseBodyEvent = { + "type": "http.response.body", + "body": await file.read(), + "more_body": False, + } + await send(body_event) + + async def start_response(self, send: Send) -> None: + """Emit the start event of the response. This event includes the headers and status codes. + + Args: + send: The ASGI send function. + + Returns: + None + """ + try: + fs_info = self.file_info = cast( + "FileInfo", (await self.file_info if iscoroutine(self.file_info) else self.file_info) + ) + except FileNotFoundError as e: + raise ImproperlyConfiguredException(f"{self.file_path} does not exist") from e + + if fs_info["type"] != "file": + raise ImproperlyConfiguredException(f"{self.file_path} is not a file") + + self.content_length = fs_info["size"] + + self.headers.setdefault("content-length", str(self.content_length)) + self.headers.setdefault("last-modified", formatdate(fs_info["mtime"], usegmt=True)) + + if self.etag: + self.headers.setdefault("etag", self.etag.to_header()) + else: + self.headers.setdefault( + "etag", + create_etag_for_file(path=self.file_path, modified_time=fs_info["mtime"], file_size=fs_info["size"]), + ) + + await super().start_response(send=send) + + +class File(Response): + """A response, streaming a file as response body.""" + + __slots__ = ( + "chunk_size", + "content_disposition_type", + "etag", + "file_path", + "file_system", + "filename", + "file_info", + "stat_result", + ) + + def __init__( + self, + path: str | PathLike | Path, + *, + background: BackgroundTask | BackgroundTasks | None = None, + chunk_size: int = ONE_MEGABYTE, + content_disposition_type: Literal["attachment", "inline"] = "attachment", + cookies: ResponseCookies | None = None, + encoding: str = "utf-8", + etag: ETag | None = None, + file_info: FileInfo | Coroutine[Any, Any, FileInfo] | None = None, + file_system: FileSystemProtocol | None = None, + filename: str | None = None, + headers: ResponseHeaders | None = None, + media_type: Literal[MediaType.TEXT] | str | None = None, + stat_result: stat_result_type | None = None, + status_code: int | None = None, + ) -> None: + """Initialize ``File`` + + Notes: + - This class extends the :class:`Stream <.response.Stream>` class. + + Args: + path: A file path in one of the supported formats. + background: A :class:`BackgroundTask <.background_tasks.BackgroundTask>` instance or + :class:`BackgroundTasks <.background_tasks.BackgroundTasks>` to execute after the response is finished. + Defaults to None. + chunk_size: The chunk sizes to use when streaming the file. Defaults to 1MB. + content_disposition_type: The type of the ``Content-Disposition``. Either ``inline`` or ``attachment``. + cookies: A list of :class:`Cookie <.datastructures.Cookie>` instances to be set under the response + ``Set-Cookie`` header. + encoding: The encoding to be used for the response headers. + etag: An optional :class:`ETag <.datastructures.ETag>` instance. If not provided, an etag will be + generated. + file_info: The output of calling :meth:`file_system.info `, equivalent to + providing an :class:`os.stat_result`. + file_system: An implementation of the :class:`FileSystemProtocol <.types.FileSystemProtocol>`. If provided + it will be used to load the file. + filename: An optional filename to set in the header. + headers: A string keyed dictionary of response headers. Header keys are insensitive. + media_type: A value for the response ``Content-Type`` header. If not provided, the value will be either + derived from the filename if provided and supported by the stdlib, or will default to + ``application/octet-stream``. + stat_result: An optional result of calling :func:os.stat:. If not provided, this will be done by the + response constructor. + status_code: An HTTP status code. + """ + + if file_system is not None and not ( + callable(getattr(file_system, "info", None)) and callable(getattr(file_system, "open", None)) + ): + raise ImproperlyConfiguredException("file_system must adhere to the FileSystemProtocol type") + + self.chunk_size = chunk_size + self.content_disposition_type = content_disposition_type + self.etag = etag + self.file_info = file_info + self.file_path = path + self.file_system = file_system + self.filename = filename or "" + self.stat_result = stat_result + + super().__init__( + content=None, + status_code=status_code, + media_type=media_type, + background=background, + headers=headers, + cookies=cookies, + encoding=encoding, + ) + + def to_asgi_response( + self, + app: Litestar | None, + request: Request, + *, + background: BackgroundTask | BackgroundTasks | None = None, + encoded_headers: Iterable[tuple[bytes, bytes]] | None = None, + cookies: Iterable[Cookie] | None = None, + headers: dict[str, str] | None = None, + is_head_response: bool = False, + media_type: MediaType | str | None = None, + status_code: int | None = None, + type_encoders: TypeEncodersMap | None = None, + ) -> ASGIFileResponse: + """Create an :class:`ASGIFileResponse ` instance. + + Args: + app: The :class:`Litestar <.app.Litestar>` application instance. + background: Background task(s) to be executed after the response is sent. + cookies: A list of cookies to be set on the response. + encoded_headers: A list of already encoded headers. + headers: Additional headers to be merged with the response headers. Response headers take precedence. + is_head_response: Whether the response is a HEAD response. + media_type: Media type for the response. If ``media_type`` is already set on the response, this is ignored. + request: The :class:`Request <.connection.Request>` instance. + status_code: Status code for the response. If ``status_code`` is already set on the response, this is + type_encoders: A dictionary of type encoders to use for encoding the response content. + + Returns: + A low-level ASGI file response. + """ + if app is not None: + warn_deprecation( + version="2.1", + deprecated_name="app", + kind="parameter", + removal_in="3.0.0", + alternative="request.app", + ) + + headers = {**headers, **self.headers} if headers is not None else self.headers + cookies = self.cookies if cookies is None else itertools.chain(self.cookies, cookies) + + media_type = self.media_type or media_type + if media_type is not None: + media_type = get_enum_string_value(media_type) + + return ASGIFileResponse( + background=self.background or background, + body=b"", + chunk_size=self.chunk_size, + content_disposition_type=self.content_disposition_type, # pyright: ignore + content_length=0, + cookies=cookies, + encoded_headers=encoded_headers, + encoding=self.encoding, + etag=self.etag, + file_info=self.file_info, + file_path=self.file_path, + file_system=self.file_system, + filename=self.filename, + headers=headers, + is_head_response=is_head_response, + media_type=media_type, + stat_result=self.stat_result, + status_code=self.status_code or status_code, + ) -- cgit v1.2.3