Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/responses.py: 21%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import hashlib
4import http.cookies
5import json
6import os
7import stat
8import sys
9from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence
10from contextlib import asynccontextmanager
11from datetime import datetime
12from email.utils import format_datetime, formatdate
13from functools import partial
14from mimetypes import guess_type
15from secrets import token_hex
16from typing import Any, Literal
17from urllib.parse import quote
19import anyio
20import anyio.to_thread
22from starlette._utils import create_collapsing_task_group
23from starlette.background import BackgroundTask
24from starlette.concurrency import iterate_in_threadpool
25from starlette.datastructures import URL, Headers, MutableHeaders
26from starlette.requests import ClientDisconnect
27from starlette.types import Message, Receive, Scope, Send
30class Response:
31 media_type = None
32 charset = "utf-8"
34 def __init__(
35 self,
36 content: Any = None,
37 status_code: int = 200,
38 headers: Mapping[str, str] | None = None,
39 media_type: str | None = None,
40 background: BackgroundTask | None = None,
41 ) -> None:
42 self.status_code = status_code
43 if media_type is not None:
44 self.media_type = media_type
45 self.background = background
46 self.body = self.render(content)
47 self.init_headers(headers)
49 def render(self, content: Any) -> bytes | memoryview:
50 if content is None:
51 return b""
52 if isinstance(content, memoryview):
53 return content.cast("B") if content.c_contiguous else content.tobytes()
54 if isinstance(content, bytes):
55 return content
56 return content.encode(self.charset) # type: ignore
58 def init_headers(self, headers: Mapping[str, str] | None = None) -> None:
59 if headers is None:
60 raw_headers: list[tuple[bytes, bytes]] = []
61 populate_content_length = True
62 populate_content_type = True
63 else:
64 raw_headers = [(k.lower().encode("latin-1"), v.encode("latin-1")) for k, v in headers.items()]
65 keys = [h[0] for h in raw_headers]
66 populate_content_length = b"content-length" not in keys
67 populate_content_type = b"content-type" not in keys
69 body = getattr(self, "body", None)
70 if (
71 body is not None
72 and populate_content_length
73 and not (self.status_code < 200 or self.status_code in (204, 304))
74 ):
75 content_length = str(len(body))
76 raw_headers.append((b"content-length", content_length.encode("latin-1")))
78 content_type = self.media_type
79 if content_type is not None and populate_content_type:
80 if content_type.startswith("text/") and "charset=" not in content_type.lower():
81 content_type += "; charset=" + self.charset
82 raw_headers.append((b"content-type", content_type.encode("latin-1")))
84 self.raw_headers = raw_headers
86 @property
87 def headers(self) -> MutableHeaders:
88 if not hasattr(self, "_headers"):
89 self._headers = MutableHeaders(raw=self.raw_headers)
90 return self._headers
92 def set_cookie(
93 self,
94 key: str,
95 value: str = "",
96 max_age: int | None = None,
97 expires: datetime | str | int | None = None,
98 path: str | None = "/",
99 domain: str | None = None,
100 secure: bool = False,
101 httponly: bool = False,
102 samesite: Literal["lax", "strict", "none"] | None = "lax",
103 partitioned: bool = False,
104 ) -> None:
105 cookie: http.cookies.BaseCookie[str] = http.cookies.SimpleCookie()
106 cookie[key] = value
107 if max_age is not None:
108 cookie[key]["max-age"] = max_age
109 if expires is not None:
110 if isinstance(expires, datetime):
111 cookie[key]["expires"] = format_datetime(expires, usegmt=True)
112 else:
113 cookie[key]["expires"] = expires
114 if path is not None:
115 cookie[key]["path"] = path
116 if domain is not None:
117 cookie[key]["domain"] = domain
118 if secure:
119 cookie[key]["secure"] = True
120 if httponly:
121 cookie[key]["httponly"] = True
122 if samesite is not None:
123 assert samesite.lower() in [
124 "strict",
125 "lax",
126 "none",
127 ], "samesite must be either 'strict', 'lax' or 'none'"
128 cookie[key]["samesite"] = samesite
129 if partitioned:
130 if sys.version_info < (3, 14):
131 raise ValueError("Partitioned cookies are only supported in Python 3.14 and above.") # pragma: no cover
132 cookie[key]["partitioned"] = True # pragma: no cover
134 cookie_val = cookie.output(header="").strip()
135 self.raw_headers.append((b"set-cookie", cookie_val.encode("latin-1")))
137 def delete_cookie(
138 self,
139 key: str,
140 path: str = "/",
141 domain: str | None = None,
142 secure: bool = False,
143 httponly: bool = False,
144 samesite: Literal["lax", "strict", "none"] | None = "lax",
145 partitioned: bool = False,
146 ) -> None:
147 self.set_cookie(
148 key,
149 max_age=0,
150 expires=0,
151 path=path,
152 domain=domain,
153 secure=secure,
154 httponly=httponly,
155 samesite=samesite,
156 partitioned=partitioned,
157 )
159 def _wrap_websocket_denial_send(self, send: Send) -> Send:
160 async def wrapped(message: Message) -> None:
161 message_type = message["type"]
162 if message_type in {"http.response.start", "http.response.body"}: # pragma: no branch
163 message = {**message, "type": "websocket." + message_type}
164 await send(message)
166 return wrapped
168 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
169 if scope["type"] == "websocket":
170 send = self._wrap_websocket_denial_send(send)
171 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
172 await send({"type": "http.response.body", "body": self.body})
174 if self.background is not None:
175 await self.background()
178class HTMLResponse(Response):
179 media_type = "text/html"
182class PlainTextResponse(Response):
183 media_type = "text/plain"
186class JSONResponse(Response):
187 media_type = "application/json"
189 def __init__(
190 self,
191 content: Any,
192 status_code: int = 200,
193 headers: Mapping[str, str] | None = None,
194 media_type: str | None = None,
195 background: BackgroundTask | None = None,
196 ) -> None:
197 super().__init__(content, status_code, headers, media_type, background)
199 def render(self, content: Any) -> bytes:
200 return json.dumps(
201 content,
202 ensure_ascii=False,
203 allow_nan=False,
204 indent=None,
205 separators=(",", ":"),
206 ).encode("utf-8")
209class RedirectResponse(Response):
210 def __init__(
211 self,
212 url: str | URL,
213 status_code: int = 307,
214 headers: Mapping[str, str] | None = None,
215 background: BackgroundTask | None = None,
216 ) -> None:
217 super().__init__(content=b"", status_code=status_code, headers=headers, background=background)
218 self.headers["location"] = quote(str(url), safe=":/%#?=@[]!$&'()*+,;")
221Content = str | bytes | memoryview
222SyncContentStream = Iterable[Content]
223AsyncContentStream = AsyncIterable[Content]
224ContentStream = AsyncContentStream | SyncContentStream
227class StreamingResponse(Response):
228 body_iterator: AsyncContentStream
230 def __init__(
231 self,
232 content: ContentStream,
233 status_code: int = 200,
234 headers: Mapping[str, str] | None = None,
235 media_type: str | None = None,
236 background: BackgroundTask | None = None,
237 ) -> None:
238 if isinstance(content, AsyncIterable):
239 self.body_iterator = content
240 else:
241 self.body_iterator = iterate_in_threadpool(content)
242 self.status_code = status_code
243 self.media_type = self.media_type if media_type is None else media_type
244 self.background = background
245 self.init_headers(headers)
247 async def listen_for_disconnect(self, receive: Receive) -> None:
248 while True:
249 message = await receive()
250 if message["type"] == "http.disconnect":
251 break
253 async def stream_response(self, send: Send) -> None:
254 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
255 async for chunk in self.body_iterator:
256 if not isinstance(chunk, bytes | memoryview):
257 chunk = chunk.encode(self.charset)
258 await send({"type": "http.response.body", "body": chunk, "more_body": True})
260 await send({"type": "http.response.body", "body": b"", "more_body": False})
262 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
263 if scope["type"] == "websocket":
264 send = self._wrap_websocket_denial_send(send)
265 await self.stream_response(send)
266 if self.background is not None:
267 await self.background()
268 return
270 spec_version = tuple(map(int, scope.get("asgi", {}).get("spec_version", "2.0").split(".")))
272 if spec_version >= (2, 4):
273 try:
274 await self.stream_response(send)
275 except OSError:
276 raise ClientDisconnect()
277 else:
278 async with create_collapsing_task_group() as task_group:
280 async def wrap(func: Callable[[], Awaitable[None]]) -> None:
281 await func()
282 task_group.cancel_scope.cancel()
284 task_group.start_soon(wrap, partial(self.stream_response, send))
285 await wrap(partial(self.listen_for_disconnect, receive))
287 if self.background is not None:
288 await self.background()
291class MalformedRangeHeader(Exception):
292 def __init__(self, content: str = "Malformed range header.") -> None:
293 self.content = content
296class RangeNotSatisfiable(Exception):
297 def __init__(self, max_size: int) -> None:
298 self.max_size = max_size
301class FileResponse(Response):
302 chunk_size = 64 * 1024
303 max_ranges = 100
305 def __init__(
306 self,
307 path: str | os.PathLike[str],
308 status_code: int = 200,
309 headers: Mapping[str, str] | None = None,
310 media_type: str | None = None,
311 background: BackgroundTask | None = None,
312 filename: str | None = None,
313 stat_result: os.stat_result | None = None,
314 content_disposition_type: str = "attachment",
315 ) -> None:
316 self.path = path
317 self.status_code = status_code
318 self.filename = filename
319 if media_type is None:
320 media_type = guess_type(filename or path)[0] or "application/octet-stream"
321 self.media_type = media_type
322 self.background = background
323 self.init_headers(headers)
324 self.headers.setdefault("accept-ranges", "bytes")
325 if self.filename is not None:
326 content_disposition_filename = quote(self.filename)
327 if content_disposition_filename != self.filename:
328 content_disposition = f"{content_disposition_type}; filename*=utf-8''{content_disposition_filename}"
329 else:
330 content_disposition = f'{content_disposition_type}; filename="{self.filename}"'
331 self.headers.setdefault("content-disposition", content_disposition)
332 self.stat_result = stat_result
333 if stat_result is not None:
334 self.set_stat_headers(stat_result)
336 def set_stat_headers(self, stat_result: os.stat_result) -> None:
337 content_length = str(stat_result.st_size)
338 last_modified = formatdate(stat_result.st_mtime, usegmt=True)
339 etag_base = str(stat_result.st_mtime) + "-" + str(stat_result.st_size)
340 etag = f'"{hashlib.md5(etag_base.encode(), usedforsecurity=False).hexdigest()}"'
342 self.headers.setdefault("content-length", content_length)
343 self.headers.setdefault("last-modified", last_modified)
344 self.headers.setdefault("etag", etag)
346 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
347 scope_type = scope["type"]
348 send_header_only = scope_type == "http" and scope["method"].upper() == "HEAD"
349 send_pathsend = scope_type == "http" and "http.response.pathsend" in scope.get("extensions", {})
350 if scope_type == "websocket":
351 send = self._wrap_websocket_denial_send(send)
353 if self.stat_result is None:
354 try:
355 stat_result = await anyio.to_thread.run_sync(os.stat, self.path)
356 self.set_stat_headers(stat_result)
357 except FileNotFoundError:
358 raise RuntimeError(f"File at path {self.path} does not exist.")
359 else:
360 mode = stat_result.st_mode
361 if not stat.S_ISREG(mode):
362 raise RuntimeError(f"File at path {self.path} is not a file.")
363 else:
364 stat_result = self.stat_result
366 headers = Headers(scope=scope)
367 http_range = headers.get("range")
368 http_if_range = headers.get("if-range")
370 if (
371 self.status_code != 200
372 or http_range is None
373 or (scope_type == "http" and scope["method"].upper() not in ("GET", "HEAD"))
374 or (http_if_range is not None and not self._should_use_range(http_if_range))
375 ):
376 send_file = partial(self._handle_simple, send, send_header_only, send_pathsend)
377 else:
378 try:
379 ranges = self._parse_range_header(http_range, stat_result.st_size)
380 except MalformedRangeHeader as exc:
381 return await PlainTextResponse(exc.content, status_code=400)(scope, receive, send)
382 except RangeNotSatisfiable as exc:
383 response = PlainTextResponse(status_code=416, headers={"Content-Range": f"bytes */{exc.max_size}"})
384 return await response(scope, receive, send)
386 if len(ranges) == 0:
387 send_file = partial(self._handle_simple, send, send_header_only, send_pathsend)
388 elif len(ranges) == 1:
389 start, end = ranges[0]
390 send_file = partial(self._handle_single_range, send, start, end, stat_result.st_size, send_header_only)
391 send_pathsend = False
392 else:
393 send_file = partial(self._handle_multiple_ranges, send, ranges, stat_result.st_size, send_header_only)
394 send_pathsend = False
396 spec_version = tuple(map(int, scope.get("asgi", {}).get("spec_version", "2.0").split(".")))
397 if scope_type != "http" or send_header_only or send_pathsend or spec_version >= (2, 4):
398 await send_file()
399 else:
400 async with create_collapsing_task_group() as task_group:
402 async def stream_file() -> None:
403 await send_file()
404 task_group.cancel_scope.cancel()
406 task_group.start_soon(stream_file)
407 while True:
408 if (await receive())["type"] == "http.disconnect":
409 task_group.cancel_scope.cancel()
410 break
412 if self.background is not None:
413 await self.background()
415 # TODO: Remove this wrapper once minimum AnyIO includes https://github.com/agronholm/anyio/pull/1314.
416 @asynccontextmanager
417 async def _open_file(self) -> AsyncIterator[anyio.AsyncFile[bytes]]:
418 file = await anyio.open_file(self.path, mode="rb")
419 try:
420 yield file
421 finally:
422 # Closing must finish even when the transfer is cancelled.
423 with anyio.CancelScope(shield=True):
424 await file.aclose()
426 async def _handle_simple(self, send: Send, send_header_only: bool, send_pathsend: bool) -> None:
427 await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
428 if send_header_only:
429 await send({"type": "http.response.body", "body": b"", "more_body": False})
430 elif send_pathsend:
431 await send({"type": "http.response.pathsend", "path": str(self.path)})
432 else:
433 async with self._open_file() as file:
434 more_body = True
435 while more_body:
436 chunk = await file.read(self.chunk_size)
437 more_body = len(chunk) == self.chunk_size
438 await send({"type": "http.response.body", "body": chunk, "more_body": more_body})
440 async def _handle_single_range(
441 self, send: Send, start: int, end: int, file_size: int, send_header_only: bool
442 ) -> None:
443 headers = MutableHeaders(raw=list(self.raw_headers))
444 headers["content-range"] = f"bytes {start}-{end - 1}/{file_size}"
445 headers["content-length"] = str(end - start)
446 await send({"type": "http.response.start", "status": 206, "headers": headers.raw})
447 if send_header_only:
448 await send({"type": "http.response.body", "body": b"", "more_body": False})
449 else:
450 async with self._open_file() as file:
451 await file.seek(start)
452 more_body = True
453 while more_body:
454 chunk = await file.read(min(self.chunk_size, end - start))
455 start += len(chunk)
456 more_body = len(chunk) == self.chunk_size and start < end
457 await send({"type": "http.response.body", "body": chunk, "more_body": more_body})
459 async def _handle_multiple_ranges(
460 self,
461 send: Send,
462 ranges: list[tuple[int, int]],
463 file_size: int,
464 send_header_only: bool,
465 ) -> None:
466 # In firefox and chrome, they use boundary with 95-96 bits entropy (that's roughly 13 bytes).
467 boundary = token_hex(13)
468 content_length, header_generator = self.generate_multipart(
469 ranges, boundary, file_size, self.headers["content-type"]
470 )
471 headers = MutableHeaders(raw=list(self.raw_headers))
472 headers["content-type"] = f"multipart/byteranges; boundary={boundary}"
473 headers["content-length"] = str(content_length)
474 await send({"type": "http.response.start", "status": 206, "headers": headers.raw})
475 if send_header_only:
476 await send({"type": "http.response.body", "body": b"", "more_body": False})
477 else:
478 async with self._open_file() as file:
479 for start, end in ranges:
480 await send({"type": "http.response.body", "body": header_generator(start, end), "more_body": True})
481 await file.seek(start)
482 while start < end:
483 chunk = await file.read(min(self.chunk_size, end - start))
484 if not chunk:
485 raise RuntimeError(f"File at path {self.path} is shorter than expected.")
486 start += len(chunk)
487 await send({"type": "http.response.body", "body": chunk, "more_body": True})
488 await send({"type": "http.response.body", "body": b"\r\n", "more_body": True})
489 await send(
490 {
491 "type": "http.response.body",
492 "body": f"--{boundary}--".encode("latin-1"),
493 "more_body": False,
494 }
495 )
497 def _should_use_range(self, http_if_range: str) -> bool:
498 if http_if_range.startswith("W/"):
499 return False
500 return http_if_range == self.headers["last-modified"] or http_if_range == self.headers["etag"]
502 @classmethod
503 def _parse_range_header(cls, http_range: str, file_size: int) -> list[tuple[int, int]]:
504 ranges: list[tuple[int, int]] = []
505 try:
506 units, range_ = http_range.split("=", 1)
507 except ValueError:
508 raise MalformedRangeHeader()
510 units = units.strip().lower()
512 if units != "bytes":
513 raise MalformedRangeHeader("Only support bytes range")
515 if range_.count(",") + 1 > cls.max_ranges:
516 return []
518 ranges = cls._parse_ranges(range_, file_size)
520 if len(ranges) == 0:
521 raise MalformedRangeHeader("Range header: range must be requested")
523 if any(not (0 <= start < file_size) for start, _ in ranges):
524 raise RangeNotSatisfiable(file_size)
526 if any(start >= end for start, end in ranges):
527 raise MalformedRangeHeader("Range header: start must be less than end")
529 if len(ranges) == 1:
530 return ranges
532 # Merge overlapping ranges
533 ranges.sort()
534 result: list[tuple[int, int]] = [ranges[0]]
535 for start, end in ranges[1:]:
536 last_start, last_end = result[-1]
537 if start <= last_end:
538 result[-1] = (last_start, max(last_end, end))
539 else:
540 result.append((start, end))
542 return result
544 @classmethod
545 def _parse_ranges(cls, range_: str, file_size: int) -> list[tuple[int, int]]:
546 ranges: list[tuple[int, int]] = []
548 for part in range_.split(","):
549 part = part.strip()
551 # If the range is empty or a single dash, we ignore it.
552 if not part or part == "-":
553 continue
555 # If the range is not in the format "start-end", we ignore it.
556 if "-" not in part:
557 continue
559 start_str, end_str = part.split("-", 1)
560 start_str = start_str.strip()
561 end_str = end_str.strip()
563 try:
564 start = int(start_str) if start_str else max(file_size - int(end_str), 0)
565 end = int(end_str) + 1 if start_str and end_str and int(end_str) < file_size else file_size
566 ranges.append((start, end))
567 except ValueError:
568 # If the range is not numeric, we ignore it.
569 continue
571 return ranges
573 def generate_multipart(
574 self,
575 ranges: Sequence[tuple[int, int]],
576 boundary: str,
577 max_size: int,
578 content_type: str,
579 ) -> tuple[int, Callable[[int, int], bytes]]:
580 r"""
581 Multipart response headers generator.
583 ```
584 --{boundary}\r\n
585 Content-Type: {content_type}\r\n
586 Content-Range: bytes {start}-{end-1}/{max_size}\r\n
587 \r\n
588 ..........content...........\r\n
589 --{boundary}\r\n
590 Content-Type: {content_type}\r\n
591 Content-Range: bytes {start}-{end-1}/{max_size}\r\n
592 \r\n
593 ..........content...........\r\n
594 --{boundary}--
595 ```
596 """
597 boundary_len = len(boundary)
598 static_header_part_len = 49 + boundary_len + len(content_type) + len(str(max_size))
599 content_length = sum(
600 (len(str(start)) + len(str(end - 1)) + static_header_part_len) # Headers
601 + (end - start) # Content
602 for start, end in ranges
603 ) + (
604 4 + boundary_len # --boundary--
605 )
606 return (
607 content_length,
608 lambda start, end: (
609 f"""\
610--{boundary}\r
611Content-Type: {content_type}\r
612Content-Range: bytes {start}-{end - 1}/{max_size}\r
613\r
614"""
615 ).encode("latin-1"),
616 )