1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
from __future__ import annotations
from os.path import commonpath
from pathlib import Path
from typing import TYPE_CHECKING, Literal, Sequence
from litestar.enums import ScopeType
from litestar.exceptions import MethodNotAllowedException, NotFoundException
from litestar.file_system import FileSystemAdapter
from litestar.response.file import ASGIFileResponse
from litestar.status_codes import HTTP_404_NOT_FOUND
__all__ = ("StaticFiles",)
if TYPE_CHECKING:
from litestar.types import Receive, Scope, Send
from litestar.types.composite_types import PathType
from litestar.types.file_types import FileInfo, FileSystemProtocol
class StaticFiles:
"""ASGI App that handles file sending."""
__slots__ = ("is_html_mode", "directories", "adapter", "send_as_attachment", "headers")
def __init__(
self,
is_html_mode: bool,
directories: Sequence[PathType],
file_system: FileSystemProtocol,
send_as_attachment: bool = False,
resolve_symlinks: bool = True,
headers: dict[str, str] | None = None,
) -> None:
"""Initialize the Application.
Args:
is_html_mode: Flag dictating whether serving html. If true, the default file will be ``index.html``.
directories: A list of directories to serve files from.
file_system: The file_system spec to use for serving files.
send_as_attachment: Whether to send the file with a ``content-disposition`` header of
``attachment`` or ``inline``
resolve_symlinks: Resolve symlinks to the directories
headers: Headers that will be sent with every response.
"""
self.adapter = FileSystemAdapter(file_system)
self.directories = tuple(Path(p).resolve() if resolve_symlinks else Path(p) for p in directories)
self.is_html_mode = is_html_mode
self.send_as_attachment = send_as_attachment
self.headers = headers
async def get_fs_info(
self, directories: Sequence[PathType], file_path: PathType
) -> tuple[Path, FileInfo] | tuple[None, None]:
"""Return the resolved path and a :class:`stat_result <os.stat_result>`.
Args:
directories: A list of directory paths.
file_path: A file path to resolve
Returns:
A tuple with an optional resolved :class:`Path <anyio.Path>` instance and an optional
:class:`stat_result <os.stat_result>`.
"""
for directory in directories:
try:
joined_path = Path(directory, file_path)
file_info = await self.adapter.info(joined_path)
if file_info and commonpath([str(directory), file_info["name"], joined_path]) == str(directory):
return joined_path, file_info
except FileNotFoundError:
continue
return None, None
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""ASGI callable.
Args:
scope: ASGI scope
receive: ASGI ``receive`` callable
send: ASGI ``send`` callable
Returns:
None
"""
if scope["type"] != ScopeType.HTTP or scope["method"] not in {"GET", "HEAD"}:
raise MethodNotAllowedException()
res = await self.handle(path=scope["path"], is_head_response=scope["method"] == "HEAD")
await res(scope=scope, receive=receive, send=send)
async def handle(self, path: str, is_head_response: bool) -> ASGIFileResponse:
split_path = path.split("/")
filename = split_path[-1]
joined_path = Path(*split_path)
resolved_path, fs_info = await self.get_fs_info(directories=self.directories, file_path=joined_path)
content_disposition_type: Literal["inline", "attachment"] = (
"attachment" if self.send_as_attachment else "inline"
)
if self.is_html_mode and fs_info and fs_info["type"] == "directory":
filename = "index.html"
resolved_path, fs_info = await self.get_fs_info(
directories=self.directories,
file_path=Path(resolved_path or joined_path) / filename,
)
if fs_info and fs_info["type"] == "file":
return ASGIFileResponse(
file_path=resolved_path or joined_path,
file_info=fs_info,
file_system=self.adapter.file_system,
filename=filename,
content_disposition_type=content_disposition_type,
is_head_response=is_head_response,
headers=self.headers,
)
if self.is_html_mode:
# for some reason coverage doesn't catch these two lines
filename = "404.html" # pragma: no cover
resolved_path, fs_info = await self.get_fs_info( # pragma: no cover
directories=self.directories, file_path=filename
)
if fs_info and fs_info["type"] == "file":
return ASGIFileResponse(
file_path=resolved_path or joined_path,
file_info=fs_info,
file_system=self.adapter.file_system,
filename=filename,
status_code=HTTP_404_NOT_FOUND,
content_disposition_type=content_disposition_type,
is_head_response=is_head_response,
headers=self.headers,
)
raise NotFoundException(
f"no file or directory match the path {resolved_path or joined_path} was found"
) # pragma: no cover
|
