/
routes.py
130 lines (104 loc) · 3.86 KB
/
routes.py
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
from __future__ import annotations
import contextlib
import dataclasses
import pathlib
import typing
import packaging.utils
from starlette.responses import Response as StarletteResponse
from starlette.responses import StreamingResponse as StarletteStreamingResponse
Content = typing.Union[bytes, str]
SyncContentStream = typing.Iterable[Content]
AsyncContentStream = typing.AsyncIterable[Content]
ContentStream = typing.Union[AsyncContentStream, SyncContentStream]
Params = typing.Mapping[str, typing.Any]
Headers = typing.Mapping[str, str]
@dataclasses.dataclass()
class Response:
content: Content = b""
status_code: int = 200
media_type: str = "text/plain"
headers: typing.Optional[Headers] = None
_http_response_class = StarletteResponse
def to_http_response(self) -> StarletteResponse:
return self._http_response_class(
content=self.content,
status_code=self.status_code,
media_type=self.media_type,
headers=self.headers,
)
@dataclasses.dataclass()
class StreamingResponse(Response):
content: ContentStream = ()
_http_response_class = StarletteStreamingResponse
@dataclasses.dataclass()
class Route:
root: pathlib.Path
to: str
async def get_page(self, params: Params) -> Response:
raise NotImplementedError()
async def get_file(self, params: Params, filename: str) -> Response:
return Response(status_code=404, content="not found")
_HTML = """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
</head>
<body>
{anchors}
</body>
</html>
"""
def _is_valid_dist_filename(filename: str) -> bool:
with contextlib.suppress(
packaging.utils.InvalidWheelFilename,
packaging.utils.InvalidVersion,
):
packaging.utils.parse_wheel_filename(filename)
return True
with contextlib.suppress(
packaging.utils.InvalidSdistFilename,
packaging.utils.InvalidVersion,
):
packaging.utils.parse_sdist_filename(filename)
return True
return False
def _iter_anchors(root: pathlib.Path) -> typing.Iterator[str]:
"""Create anchor tags from directory listing.
Files are served at "{prefix}/{project}/{filename}". Entries that do not
look like a distribution file are ignored.
"""
for path in root.iterdir():
if not _is_valid_dist_filename(path.name):
continue
yield f'<a href="./{path.name}">{path.name}</a>'
class PathRoute(Route):
async def get_page(self, params: Params) -> Response:
path = self.root.joinpath(self.to.format(**params))
if path.is_file():
return Response(content=path.read_bytes(), media_type="text/html")
if path.is_dir():
html = _HTML.format(anchors="\n".join(_iter_anchors(path)))
return Response(content=html, media_type="text/html")
return Response(status_code=404, content="Not Found")
async def get_file(self, params: Params, filename: str) -> Response:
path = self.root.joinpath(self.to.format(**params))
if not path.is_dir():
return await super().get_file(params, filename)
path = path.joinpath(filename)
if not path.is_file():
return await super().get_file(params, filename)
if not _is_valid_dist_filename(path.name):
return await super().get_file(params, filename)
if filename.endswith(".tar.gz"):
media_type = "application/x-tar"
elif path.suffix in (".whl", ".zip"):
media_type = "application/zip"
else:
media_type = "application/octet-stream"
data = path.read_bytes()
return Response(status_code=200, content=data, media_type=media_type)
class HTTPRoute(Route):
async def get_page(self, params: Params) -> Response:
url = self.to.format(**params)
return Response(status_code=302, headers={"Location": url})