(
self,
send: Send,
ranges: list[tuple[int, int]],
file_size: int,
send_header_only: bool,
)
| 415 | await send({class="st">"type": class="st">"http.response.body", class="st">"body": chunk, class="st">"more_body": more_body}) |
| 416 | |
| 417 | async def _handle_multiple_ranges( |
| 418 | self, |
| 419 | send: Send, |
| 420 | ranges: list[tuple[int, int]], |
| 421 | file_size: int, |
| 422 | send_header_only: bool, |
| 423 | ) -> None: |
| 424 | class="cm"># In firefox and chrome, they use boundary with 95-96 bits entropy (that's roughly 13 bytes). |
| 425 | boundary = token_hex(13) |
| 426 | content_length, header_generator = self.generate_multipart( |
| 427 | ranges, boundary, file_size, self.headers[class="st">"content-type"] |
| 428 | ) |
| 429 | headers = MutableHeaders(raw=list(self.raw_headers)) |
| 430 | headers[class="st">"content-type"] = fclass="st">"multipart/byteranges; boundary={boundary}" |
| 431 | headers[class="st">"content-length"] = str(content_length) |
| 432 | await send({class="st">"type": class="st">"http.response.start", class="st">"status": 206, class="st">"headers": headers.raw}) |
| 433 | if send_header_only: |
| 434 | await send({class="st">"type": class="st">"http.response.body", class="st">"body": bclass="st">"", class="st">"more_body": False}) |
| 435 | else: |
| 436 | async with await anyio.open_file(self.path, mode=class="st">"rb") as file: |
| 437 | for start, end in ranges: |
| 438 | await send({class="st">"type": class="st">"http.response.body", class="st">"body": header_generator(start, end), class="st">"more_body": True}) |
| 439 | await file.seek(start) |
| 440 | while start < end: |
| 441 | chunk = await file.read(min(self.chunk_size, end - start)) |
| 442 | start += len(chunk) |
| 443 | await send({class="st">"type": class="st">"http.response.body", class="st">"body": chunk, class="st">"more_body": True}) |
| 444 | await send({class="st">"type": class="st">"http.response.body", class="st">"body": bclass="st">"\r\n", class="st">"more_body": True}) |
| 445 | await send( |
| 446 | { |
| 447 | class="st">"type": class="st">"http.response.body", |
| 448 | class="st">"body": fclass="st">"--{boundary}--".encode(class="st">"latin-1"), |
| 449 | class="st">"more_body": False, |
| 450 | } |
| 451 | ) |
| 452 | |
| 453 | def _should_use_range(self, http_if_range: str) -> bool: |
| 454 | return http_if_range == self.headers[class="st">"last-modified"] or http_if_range == self.headers[class="st">"etag"] |
no test coverage detected